#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Bybit 주식/ETF 퍼프 신규 상장 감지 → 텔레그램 import json, os, re, urllib.request, urllib.parse, sys, datetime BASE = os.path.expanduser("~/Claude/Projects/에이전트구축") SNAP = os.path.join(BASE, "bybit_stocks.json") NQLEV = os.path.join(BASE, "nq_levels.py") DRY = "--dry" in sys.argv # 뜨면 바로 알려야 할 것들 (ETF 대체재 우선) PRIORITY = ("SOXX","SMH","XSD","PSI","FTXL","XLK","QQQ","IGV","VGT","IYW", "SKYY","AIQ","BOTZ","ARKK","EWY","EWT","EWJ") # 관심 섹터 유니버스 (신규상장이 여기 걸리면 별도 알림) SECTORS = { "메모리": {"MU","SNDK","WDC","STX","SKHY","MICRON","NANYA","KIOX","KIOXIA","SIMO","RMBS","NLST"}, "광학/CPO": {"AAOI","LITE","COHR","CIEN","GLW","CRDO","ALAB","MRVL","VIAV","FN","INFN","POET","ANET","NPTN"}, "반도체장비": {"AMAT","LRCX","KLAC","TER","ASML","AEHR","ONTO","NOVA","FORM","ACLS","UCTT","ICHR","CAMT","NVMI","AXTI","AMKR","ASX"}, "AI인프라/네오클라우드": {"NBIS","CRWV","IREN","SMCI","DELL","HPE","APLD","CORZ","WULF"}, "전력": {"BE","GEV","VRT","FLNC","VICR","MPWR","POWI","NVTS","WOLF","ON","ALGM","STM","ETN","PWR","CEG","VST","TLN","OKLO","SMR","NNE","NEE"}, "방산/우주": {"PLTR","RDW","ONDS","ASTS","RKLB","SPCX","LMT","NOC","RTX","LHX","KTOS","AVAV"}, } def send(msg): s = open(NQLEV, encoding="utf-8").read() tok = re.search(r'BOT_TOKEN\s*=\s*["\']([^"\']+)', s).group(1) cid = re.search(r'CHAT_ID\s*=\s*["\']?([^"\'\s]+)', s).group(1) d = urllib.parse.urlencode({"chat_id": cid, "text": msg}).encode() urllib.request.urlopen(f"https://api.telegram.org/bot{tok}/sendMessage", d, timeout=15) # ── 신규 상장분 정체 파악용 (티커만으론 못 알아봄: APPSTOCK=AppLovin, MUU=MU 2x 등) ALIAS = {"AMDSTOCK": "AMD", "CATSTOCK": "CAT", "APPSTOCK": "APP", "SKHYNIX": "000660.KS", "SAMSUNG": "005930.KS", "HYUNDAI": "005380.KS"} LEVPAT = re.compile(r"(bull|bear|2x|3x|daily|tradr|ultra|direxion|proshares|granite|leverag)", re.I) TECH_SEC = {"Technology", "Communication Services"} KW = [("Health|Pharma|Bio|Medic|Drug", "Healthcare"), ("Bank|Financ|Capital|Insur|Broker", "Financial Services"), ("Energy|Oil|Gas|Petro", "Energy"), ("Gold|Silver|Mining|Materials", "Basic Materials"), ("Real Estate|REIT", "Real Estate"), ("Utilit", "Utilities"), ("Consumer|Retail|Staples", "Consumer Defensive"), ("Industrial|Aerospace|Defense", "Industrials")] SECT_ETF = {"Healthcare": "XLV", "Financial Services": "XLF", "Energy": "XLE", "Industrials": "XLI", "Consumer Defensive": "XLP", "Consumer Cyclical": "XLY", "Basic Materials": "XLB", "Real Estate": "XLRE", "Utilities": "XLU"} def _perf(t): try: u = ("https://query1.finance.yahoo.com/v8/finance/chart/" + urllib.parse.quote(t) + "?interval=1d&range=1y") r = urllib.request.Request(u, headers={"User-Agent": "Mozilla/5.0"}) d = json.load(urllib.request.urlopen(r, timeout=20))["chart"]["result"][0] q = d["indicators"]["quote"][0]["close"] ts = [datetime.datetime.fromtimestamp(a, datetime.timezone.utc).date() for a in d["timestamp"]] c = [x for x in q if x] last = c[-1] ytd = [v for dt, v in zip(ts, q) if v and dt >= datetime.date(datetime.date.today().year, 1, 1)] f = lambda k: (last / c[-k] - 1) * 100 if len(c) > k else None nm = d["meta"].get("longName") or d["meta"].get("shortName") or t return nm, last, ((last / ytd[0] - 1) * 100 if ytd else None), f(63), f(21) except Exception: return None, None, None, None, None def _sector(t, nm): try: import yfinance as yf sec = yf.Ticker(t).info.get("sector") if sec: return sec except Exception: pass if nm: for pat, sec in KW: if re.search(pat, nm, re.I): return sec return None def profile(sym): base = sym.replace("USDT", "") tk = ALIAS.get(base, base) nm, px, ytd, m3, m1 = _perf(tk) if not nm: return None return dict(sym=sym, tk=tk, name=nm, px=px, ytd=ytd, m3=m3, m1=m1, lev=bool(LEVPAT.search(nm)), sec=_sector(tk, nm)) def nontech_alert(new_syms): rows = [profile(s) for s in new_syms] rows = [r for r in rows if r] hit = [r for r in rows if not r["lev"] and r["sec"] and r["sec"] not in TECH_SEC and ((r["m3"] or -99) > 0 or (r["m1"] or -99) > 0)] if not hit: return None fmt = lambda v: f"{v:+.1f}%" if v is not None else "n/a" out = ["[신규상장 비테크 주목 " + str(datetime.date.today()) + "]"] for r in hit: out.append("") out.append(f"{r['sym'].replace('USDT','')} — {r['name']}") out.append(f"{r['sec']} / 현재 {r['px']:,.2f}") out.append(f"YTD {fmt(r['ytd'])} · 3M {fmt(r['m3'])} · 1M {fmt(r['m1'])}") etf = SECT_ETF.get(r["sec"]) if etf: _, _, ey, e3, e1 = _perf(etf) out.append(f"섹터({etf}) YTD {fmt(ey)} · 3M {fmt(e3)} · 1M {fmt(e1)}") skip = [r["sym"].replace("USDT", "") for r in rows if r not in hit] if skip: out.append("") out.append("제외(테크/레버리지/추세미달): " + ", ".join(skip)) return "\n".join(out) def sector_alert(new_syms): hits = [] for sy in new_syms: b = re.sub(r"STOCK$", "", sy.replace("USDT", "")) for sec, tks in SECTORS.items(): if b in tks: hits.append((sec, b)); break if not hits: return None fmt = lambda v: f"{v:+.1f}%" if v is not None else "n/a" out = ["[★ 관심섹터 신규상장 " + str(datetime.date.today()) + "]"] for sec, b in hits: nm, px, ytd, m3, m1 = _perf(ALIAS.get(b, b)) out.append("") out.append(f"{b} — {nm or '이름 확인불가'} [{sec}]") if px: out.append(f"현재 {px:,.2f}") out.append(f"YTD {fmt(ytd)} · 3M {fmt(m3)} · 1M {fmt(m1)}") out.append("") out.append("→ 유동성 3중필터 확인 필요: 거래대금 100만불+ / 뎁스 2만불+ / 매수매도 비대칭 2.5배 이하 / 스프레드 0.3% 이하") return "\n".join(out) u = "https://api.bybit.com/v5/market/instruments-info?category=linear&limit=1000" req = urllib.request.Request(u, headers={"User-Agent": "Mozilla/5.0"}) d = json.load(urllib.request.urlopen(req, timeout=20)) cur = sorted({x["symbol"] for x in d["result"]["list"] if x.get("symbolType") == "stock"}) try: old = json.load(open(SNAP, encoding="utf-8")) prev = sorted({x["symbol"] if isinstance(x, dict) else x for x in (old.get("list") if isinstance(old, dict) else old)}) except Exception: prev = [] new = [s for s in cur if s not in prev] gone = [s for s in prev if s not in cur] if not DRY: json.dump({"date": str(datetime.date.today()), "count": len(cur), "list": cur}, open(SNAP, "w", encoding="utf-8"), ensure_ascii=False, indent=1) if not prev: print(f"기준 스냅샷 생성: {len(cur)}개"); sys.exit() lines = [] if new: hot = [s for s in new if s.replace("USDT","").replace("STOCK","") in PRIORITY] if hot: lines.append("★ 관심 ETF 상장: " + ", ".join(hot)) lines.append("신규 " + str(len(new)) + "건: " + ", ".join(new)) if gone: lines.append("상장폐지 " + str(len(gone)) + "건: " + ", ".join(gone)) if lines: msg = f"[Bybit 종목 변동 {datetime.date.today()}] 총 {len(cur)}개\n" + "\n".join(lines) print(msg) if not DRY: send(msg) if new: try: ms = sector_alert(new) except Exception as e: ms = None; print("[!] 섹터알림 실패:", e) if ms: print("\n" + ms) if not DRY: send(ms) try: m2 = nontech_alert(new) except Exception as e: m2 = None; print("[!] 프로파일 실패:", e) if m2: print("\n" + m2) if not DRY: send(m2) else: print(f"변동 없음 (총 {len(cur)}개)")