""" interactive_bot.py — Telegram Bot ตอบคำถาม (long polling) ========================================================== รันแบบ foreground: python3 interactive_bot.py หรือรันเป็น service (systemd) เพื่อให้คอยฟังข้อความตลอดเวลา """ import os import re import sys import time import threading import requests import yfinance as yf import pandas as pd from datetime import datetime from config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID from scanner_common import ( calculate_rsi, drop_incomplete_last_bar, rolling_high, ) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) ALLOWED_CHAT_ID = TELEGRAM_CHAT_ID # ========================================================= # UNIVERSE # ========================================================= UNIVERSE_THAI = [ "PTT.BK", "PTTEP.BK", "TOP.BK", "BCP.BK", "GULF.BK", "GPSC.BK", "BGRIM.BK", "BANPU.BK", "KBANK.BK", "SCB.BK", "BBL.BK", "KTB.BK", "TTB.BK", "TISCO.BK", "SAWAD.BK", "MTC.BK", "TIDLOR.BK", "CPALL.BK", "CPAXT.BK", "CRC.BK", "BJC.BK", "HMPRO.BK", "GLOBAL.BK", "COM7.BK", "ADVANC.BK", "TRUE.BK", "DELTA.BK", "HANA.BK", "KCE.BK", "BDMS.BK", "BH.BK", "BCH.BK", "CHG.BK", "AOT.BK", "BEM.BK", "BTS.BK", "CPN.BK", "AP.BK", "SPALI.BK", "LH.BK", "SIRI.BK", "SCC.BK", "SCGP.BK", "WHA.BK", "AMATA.BK", "CPF.BK", "TU.BK", "CBG.BK", "OSP.BK", "ICHI.BK", "ITC.BK", "AAI.BK", "MINT.BK", "CENTEL.BK", "ERW.BK", "IVL.BK", "PTTGC.BK" ] SECTOR_MAP_THAI = { "PTT.BK": "พลังงาน", "PTTEP.BK": "สำรวจปิโตรเลียม", "TOP.BK": "โรงกลั่น", "BCP.BK": "พลังงาน", "GULF.BK": "ไฟฟ้า", "GPSC.BK": "ไฟฟ้า", "BGRIM.BK": "ไฟฟ้า", "BANPU.BK": "ถ่านหิน/พลังงาน", "KBANK.BK": "ธนาคาร", "SCB.BK": "ธนาคาร", "BBL.BK": "ธนาคาร", "KTB.BK": "ธนาคาร", "TTB.BK": "ธนาคาร", "TISCO.BK": "เช่าซื้อ", "SAWAD.BK": "สินเชื่อ", "MTC.BK": "สินเชื่อ", "TIDLOR.BK": "สินเชื่อ", "CPALL.BK": "ค้าปลีก", "CPAXT.BK": "ค้าปลีก", "CRC.BK": "ค้าปลีก", "BJC.BK": "ค้าปลีก", "HMPRO.BK": "ค้าปลีก", "GLOBAL.BK": "ค้าปลีก", "COM7.BK": "ค้าปลีก IT", "ADVANC.BK": "โทรคมนาคม", "TRUE.BK": "โทรคมนาคม", "DELTA.BK": "อิเล็กทรอนิกส์", "HANA.BK": "เซมิคอนดักเตอร์", "KCE.BK": "PCB", "BDMS.BK": "โรงพยาบาล", "BH.BK": "โรงพยาบาล", "BCH.BK": "โรงพยาบาล", "CHG.BK": "โรงพยาบาล", "AOT.BK": "สนามบิน", "BEM.BK": "รถไฟฟ้า/ทางด่วน", "BTS.BK": "รถไฟฟ้า", "CPN.BK": "ศูนย์การค้า", "AP.BK": "อสังหา", "SPALI.BK": "อสังหา", "LH.BK": "อสังหา", "SIRI.BK": "อสังหา", "SCC.BK": "วัสดุก่อสร้าง", "SCGP.BK": "บรรจุภัณฑ์", "WHA.BK": "นิคมอุตสาหกรรม", "AMATA.BK": "นิคมอุตสาหกรรม", "CPF.BK": "อาหาร", "TU.BK": "อาหารทะเล", "CBG.BK": "เครื่องดื่ม", "OSP.BK": "เครื่องดื่ม", "ICHI.BK": "เครื่องดื่ม", "ITC.BK": "อาหารสัตว์", "AAI.BK": "อาหารสัตว์", "MINT.BK": "โรงแรม/อาหาร", "CENTEL.BK": "โรงแรม/อาหาร", "ERW.BK": "โรงแรม", "IVL.BK": "ปิโตรเคมี", "PTTGC.BK": "ปิโตรเคมี" } # ========================================================= # TELEGRAM HELPERS # ========================================================= def send_telegram(chat_id, text, parse_mode="Markdown"): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" payload = { "chat_id": chat_id, "text": text, "parse_mode": parse_mode, "disable_web_page_preview": True, } try: res = requests.post(url, data=payload, timeout=10) return res.json() except Exception as e: print(f"Error sending telegram: {e}") return None def send_typing(chat_id): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendChatAction" payload = {"chat_id": chat_id, "action": "typing"} try: requests.post(url, json=payload, timeout=5) except Exception: pass def authorized(chat_id): return str(chat_id) == str(ALLOWED_CHAT_ID) # ========================================================= # DATA FETCH # ========================================================= def get_stock_info(symbol): """ดึงราคา + RSI + drawdown ของหุ้น""" t = yf.Ticker(symbol) df = t.history(period="1y") df, _ = drop_incomplete_last_bar(df, "TH") if df.empty or len(df) < 30: return None latest_close = df['Close'].iloc[-1] df['52W_High'] = rolling_high(df, window=250, min_periods=30) high_52w = df['52W_High'].iloc[-1] drawdown = ((latest_close - high_52w) / high_52w) * 100 df['RSI'] = calculate_rsi(df['Close']) latest_rsi = df['RSI'].iloc[-1] prev_close = df['Close'].iloc[-2] chg = latest_close - prev_close chg_pct = (chg / prev_close) * 100 ema9 = df['Close'].ewm(span=9, adjust=False).mean().iloc[-1] ema50 = df['Close'].ewm(span=50, adjust=False).mean().iloc[-1] return { "symbol": symbol, "price": latest_close, "chg": chg, "chg_pct": chg_pct, "rsi": latest_rsi, "drawdown": drawdown, "ema9": ema9, "ema50": ema50, } def resolve_symbol(text): """แปลงคำถาม → ticker Yahoo Finance""" text = text.strip().upper().replace(" ", "") if text.endswith(".BK"): return text if text in UNIVERSE_THAI else None if text in [s.replace(".BK", "") for s in UNIVERSE_THAI]: return text + ".BK" return None # ========================================================= # TECHNICAL ANALYSIS (rule-based, ไม่ใช้ AI) # ========================================================= def ask_ai(symbol, info): """วิเคราะห์ตามสัญญาณเทคนิค (rule-based) — ผลคงที่ทุกครั้ง""" rsi = info['rsi'] dd = abs(info['drawdown']) price = info['price'] ema9 = info['ema9'] ema50 = info['ema50'] # แนวโน้มจาก EMA if price > ema9 and price > ema50: trend = "แนวโน้มขึ้น — ยืนเหนือ EMA9/EMA50" elif price > ema50: trend = "เริ่มฟื้นตัว — เหนือ EMA50 แต่ยังใต้ EMA9" elif price > ema9: trend = "เด้งสั้น — เหนือ EMA9 แต่ยังใต้ EMA50" else: trend = "แนวโน้มลง — อยู่ใต้ EMA9/EMA50" # โซน RSI if rsi < 30: rsi_msg = "RSI ต่ำมาก (oversold) เข้าโซนของถูก" elif rsi < 40: rsi_msg = "RSI เริ่มต่ำ มีโอกาสกลับตัว" elif rsi < 50: rsi_msg = "RSI ปานกลาง ยังไม่สุดทาง" elif rsi <= 70: rsi_msg = "RSI ปกติ" else: rsi_msg = "RSI สูง (overbought) ระวังย่อตัว" support = min(ema9, ema50) resist = max(ema9, ema50) return (f"• {trend}\n" f"• {rsi_msg} (ย่อจาก High 52W {dd:.1f}%)\n" f"• แนวรับอ้างอิง ฿{support:.2f} | แนวต้าน ฿{resist:.2f}") # ========================================================= # COMMAND HANDLERS # ========================================================= def cmd_start(chat_id): text = ( "🤖 *StockBotKit* — หุ้นไทย SET100\n\n" "คำสั่งที่ใช้ได้:\n" "• `/start` — เมนูนี้\n" "• `/scan` — สแกนหุ้น SET100 ทันที\n" "• `/price [ชื่อ]` — ดูราคาหุ้น เช่น `/price PTT`\n" "• `/help` — วิธีใช้\n\n" "💡 พิมพ์ชื่อหุ้นตรงๆ ก็ได้ เช่น `KBANK`\n" "ขยันถามได้ ไม่จำกัด" ) send_telegram(chat_id, text) def cmd_help(chat_id): text = ( "📖 *วิธีใช้ StockBotKit*\n\n" "• `/scan` — สแกนหุ้น SET100 ผลลัพธ์จะส่งกลับที่นี่\n" "• `/price PTT` — ดูราคา + RSI + AI วิเคราะห์\n" "• พิมพ์ `KBANK` — เหมือน `/price`\n" "• `/status` — สถานะเครื่อง\n\n" "เพียงเท่านี้เองครับ! ระบบจะสแกนเองทุกวัน 17:00\n" "และส่งแจ้งเตือนหุ้นน่าสนใจมาให้อัตโนมัติ" ) send_telegram(chat_id, text) def cmd_status(chat_id): text = "⚙️ *สถานะระบบ*\n" try: with open('/proc/meminfo', 'r') as mem: lines = mem.readlines() total = int(lines[0].split()[1]) / 1024 / 1024 avail = int(lines[2].split()[1]) / 1024 / 1024 used = total - avail text += f"• RAM: {used:.1f}GB / {total:.1f}GB\n" except Exception: pass try: st = os.statvfs('/') disk_total = (st.f_blocks * st.f_frsize) / (1024**3) disk_avail = (st.f_bavail * st.f_frsize) / (1024**3) text += f"• Disk: {disk_total - disk_avail:.0f}GB / {disk_total:.0f}GB\n" except Exception: pass text += f"• วิเคราะห์: rule-based (ไม่ใช้ AI)\n" text += f"• โฟลเดอร์: {BASE_DIR}\n" text += f"• Last check: {datetime.now().strftime('%d/%m/%Y %H:%M')}" send_telegram(chat_id, text) def cmd_scan(chat_id): send_typing(chat_id) send_telegram(chat_id, "🔎 กำลังสแกนหุ้น SET100 กรุณารอสักครู่...") import subprocess try: result = subprocess.run( [sys.executable, os.path.join(BASE_DIR, "stock_bot_thai.py")], capture_output=True, text=True, timeout=600) print(result.stdout[-2000:]) if result.returncode == 0: send_telegram(chat_id, "✅ *สแกนเสร็จสิ้น!*\nผลลัพธ์ถูกบันทึกที่ dashboard.html\nถ้ามีหุ้นเกรด A+ จะส่งรายละเอียดให้ตามข้างบนครับ") else: err = result.stderr[-500:] if result.stderr else "unknown error" send_telegram(chat_id, f"❌ สแกน error:\n```\n{err}\n```", parse_mode="Markdown") except subprocess.TimeoutExpired: send_telegram(chat_id, "⏰ สแกนใช้เวลานานเกิน 10 นาที อาจมีปัญหา network") def cmd_price(chat_id, symbol): send_typing(chat_id) info = get_stock_info(symbol) if info is None: send_telegram(chat_id, f"❌ ไม่พบข้อมูลหุ้น `{symbol}` หรือมีปัญหาโหลดข้อมูล", parse_mode="Markdown") return emoji = "🟢" if info['chg'] >= 0 else "🔴" clean = symbol.replace(".BK", "") sector = SECTOR_MAP_THAI.get(symbol, "") chg_str = f"{info['chg']:+.2f} ({info['chg_pct']:+.2f}%)" text = ( f"{emoji} *{clean}* — {sector}\n" f"ราคา: *฿{info['price']:.2f}* ({chg_str})\n" f"RSI(14): *{info['rsi']:.1f}* " f"{'⚠️ Oversold' if info['rsi'] < 30 else ('⚠️ Overbought' if info['rsi'] > 70 else 'ปกติ')}\n" f"ย่อตัวจาก High 52W: *{-abs(info['drawdown']):.1f}%*\n" f"EMA9: {info['ema9']:.2f} | EMA50: {info['ema50']:.2f}" ) send_telegram(chat_id, text) ai_text = ask_ai(symbol, info) if ai_text: send_telegram(chat_id, f"🧠 *AI วิเคราะห์:*\n{ai_text}") else: send_telegram(chat_id, "_(AI คิดไม่ออกตอนนี้ ลองถามใหม่ทีหลังนะครับ)_") # ========================================================= # MAIN LOOP (LONG POLLING) # ========================================================= def get_updates(offset): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getUpdates" params = {"timeout": 30, "offset": offset} try: res = requests.get(url, params=params, timeout=35) data = res.json() if data.get("ok"): return data["result"] except Exception as e: print(f"getUpdates error: {e}") return [] def handle_message(chat_id, text): if not authorized(chat_id): send_telegram(chat_id, "⚠️ คุณไม่ได้รับอนุญาตให้ใช้ bot นี้ครับ") return text = text.strip() cmd = text.split()[0].lower() if text else "" if cmd == "/start": cmd_start(chat_id) elif cmd == "/help": cmd_help(chat_id) elif cmd == "/status": cmd_status(chat_id) elif cmd == "/scan": cmd_scan(chat_id) elif cmd == "/price": parts = text.split(maxsplit=1) if len(parts) < 2: send_telegram(chat_id, "พิมพ์ชื่อหุ้นด้วยครับ เช่น `/price KBANK`", parse_mode="Markdown") return symbol = resolve_symbol(parts[1]) if symbol: cmd_price(chat_id, symbol) else: send_telegram(chat_id, f"❌ ไม่รู้จักหุ้น `{parts[1]}` — พิมพ์ `/help` ดูวิธีใช้", parse_mode="Markdown") else: # พิมพ์ชื่อหุ้นตรงๆ symbol = resolve_symbol(text) if symbol: cmd_price(chat_id, symbol) else: send_telegram( chat_id, "❓ ไม่เข้าใจคำสั่งครับ\nพิมพ์ `/help` เพื่อดูคำสั่งที่ใช้ได้", parse_mode="Markdown") def main(): print("🚀 StockBotKit Telegram Bot เริ่มทำงาน...") print(f" Bot token: ...{TELEGRAM_BOT_TOKEN[-6:]}") print(f" Chat ID: {TELEGRAM_CHAT_ID}") print(" วิเคราะห์: rule-based (ไม่ใช้ AI)") print("-" * 50) offset = 0 while True: try: updates = get_updates(offset) for u in updates: if "message" in u: msg = u["message"] chat_id = msg["chat"]["id"] text = msg.get("text", "") if text: print(f"📩 [{datetime.now().strftime('%H:%M:%S')}] chat={chat_id}: {text[:60]}") handle_message(chat_id, text) offset = u["update_id"] + 1 except KeyboardInterrupt: print("\n👋 หยุด bot") break except Exception as e: print(f"⚠️ Unexpected error: {e}") time.sleep(3) if __name__ == "__main__": main()