이미 급등한 알트코인 말고, 아직 아래에 있는 종목을 찾는 방법 — 업비트 Envelope 검색기 직접 만들기
최근 비트코인의 단기 상승세가 강해지면서 알트코인까지 매수세가 확산되는 모습이 나타나고 있습니다.
특정 알트코인 몇 개만 움직이는 것이 아니라 여러 종목이 돌아가면서 상승하는 이른바 순환매 형태의 단기 소불장에 가까운 모습입니다.
실제로 최근 시장에서도 비트코인이 8월 6만 달러 아래에서 9월 8만 달러대로 반등한 뒤 알트코인의 시장 참여 폭이 확대됐다는 분석들이 나오고 있습니다. 다만 모든 지표가 완전한 알트 시즌을 가리키는 것은 아니기 때문에, 개인적으로는 아직은 “비트코인 상승 이후 알트코인으로 매수세가 확산되는 구간” 정도로 보고 있습니다.
이럴 때 단타를 하는 방법도 있지만, 저는 조금 다른 검색기가 하나 있으면 좋겠다는 생각이 들었습니다.
이미 슈팅 나온 코인을 굳이 따라가야 할까?
예를 들어 하루에 이미 20%, 30%, 50% 올라버린 코인이 있습니다.
물론 여기서 다시 50%가 올라갈 수도 있습니다.
하지만 매수하는 입장에서는 이미 변동성이 크게 확대된 종목을 따라가는 것보다 아직 가격이 낮은 곳에 있으면서 이제 막 추세를 돌리려는 종목을 찾는 방법도 있습니다.
그래서 아주 단순한 보조지표 하나를 이용했습니다.
바로 Envelope(엔벨로프)입니다.
60일 Envelope ±8%
이번 검색기에서 사용하는 기본 설정은 다음과 같습니다.
60일 이동평균선 ±8%
따라서 하단선은 간단합니다.
ENV 하단 = 60일 이동평균 × 0.92
복잡한 AI 모델도 아니고 수십 개의 보조지표를 조합한 것도 아닙니다.
그냥 이것뿐입니다.
가격이 충분히 아래쪽에 내려와 있는 종목을 찾고, 그 종목이 다시 Envelope 하단선을 아래에서 위로 돌파하는 순간을 검색합니다.
예를 들어 이런 형태입니다.
시가 < ENV 하단
종가 > ENV 하단
오늘 진행 중인 봉이라면 종가 대신 현재 가격을 사용합니다.
즉,
오늘 시가 < ENV 하단
현재가 > ENV 하단
이면 0봉전 돌파로 검색됩니다.
왜 하단선 아래에 있는 종목을 찾나?
목적은 간단합니다.
이미 크게 올라간 종목을 찾아내는 검색기가 아닙니다.
아직 상대적으로 낮은 가격대에 있는 종목을 먼저 추려내기 위한 검색기입니다.
그리고 단순히 Envelope 아래에 있다는 이유만으로 매수하는 것도 아닙니다.
계속 떨어지는 종목은 Envelope 하단 아래에서 며칠이고 머물 수 있기 때문입니다.
그래서 두 가지를 나눠서 봅니다.
① 아직 ENV 하단 아래에 있는 종목
그리고
② ENV 하단 아래에 있다가 다시 위로 올라온 종목
입니다.
두 번째 조건은 적어도 가격이 하단선을 회복하려는 움직임이 발생했다는 뜻입니다.
물론 이것만으로 상승 추세가 확정됐다는 의미는 아닙니다.
보조지표 하나로 미래 가격을 알 수는 없습니다.
매수와 손절 기준도 단순하게
이걸 중기 추세 전략으로 활용한다면 구조 역시 복잡하게 만들 필요가 없습니다.
예를 들어 저는 이런 형태를 생각할 수 있습니다.
ENV 하단을 상향 돌파한 종목을 편입하고, 돌파 기준이 됐던 가격대를 다시 명확하게 깨면 손절합니다.
그러면 구조는 상당히 단순해집니다.
낮은 가격대 검색 → 하단선 회복 → 편입 → 추세 유지 → 기준선 이탈 시 손절
단타처럼 몇 틱을 먹고 바로 나오는 방식이 아니라, 시장에 순환매가 계속 이어진다는 가정 아래 며칠에서 조금 더 긴 중기 추세를 가져가 보기 위한 검색 도구에 가깝습니다.
중요한 것은 Envelope가 특별한 지표라서 이 전략을 사용하는 것이 아니라는 점입니다.
Envelope가 중요한 게 아니다
사실 이번 검색기를 만든 가장 큰 이유는 이것입니다.
자기가 사용하는 매매 기준이 있다면 검색기는 누구나 만들 수 있습니다.
저는 이번에 60일 Envelope 8%를 사용했을 뿐입니다.
누군가는 20일 이동평균선을 사용할 수도 있고,
RSI를 사용할 수도 있고,
볼린저밴드 하단 이탈 후 복귀를 사용할 수도 있고,
120일 이동평균선이나 자신이 직접 만든 지표를 사용할 수도 있습니다.
결국 필요한 것은
“내가 차트를 하나씩 열어보면서 찾고 있는 조건을 코드로 표현할 수 있는가?”
입니다.
조건을 숫자로 정의할 수 있다면 컴퓨터가 업비트 전 종목을 대신 검사하게 만들 수 있습니다.
사람이 업비트 종목 200개를 하나씩 열어볼 필요가 없습니다.
0봉전부터 과거 N봉까지 검색
검색기는 현재 봉만 검색하지 않습니다.
GUI에서
0봉전 ~ 5봉전
처럼 설정할 수 있도록 만들었습니다.
여기서
- 0봉전 = 오늘 진행 중인 일봉
- 1봉전 = 어제 완성된 일봉
- 2봉전 = 이틀 전
- 3봉전 = 3일 전
이라는 뜻입니다.
예를 들어 0 ~ 5로 설정하면 최근 6개의 일봉을 검사해서 어느 날 Envelope 하단 상향돌파가 발생했는지 찾아냅니다.
검색 결과에도 0봉전, 2봉전처럼 실제 돌파 시점을 표시합니다.
여기서 중요한 부분이 하나 있습니다.
과거 3봉 전의 돌파를 검색하면서 오늘의 60일 이동평균을 사용하는 것이 아닙니다.
3봉 전이면 3봉 전 당시의 60일 이동평균과 Envelope 하단을 다시 계산해서 돌파 여부를 판단하도록 만들었습니다.
거래량도 같이 확인한다
가격만 보는 것보다 거래량을 함께 보면 종목을 거르는 데 도움이 됩니다.
그래서 검색 결과에 다음 데이터도 같이 표시하도록 했습니다.
현재가 / 60일 MA / ENV 하단 / 하단 대비 이격률 / 24시간 거래대금 / 오늘 거래량 / 최근 7일 평균 거래량 / 7일 평균 대비 현재 거래량
예를 들어 최근 7개 완성 일봉의 하루 평균 거래량이 100만 개인데 오늘 거래량이 250만 개라면,
7일 평균 대비 거래량 = 250%
로 표시됩니다.
가격이 하단선을 돌파했는데 거래량까지 평소보다 크게 증가하고 있다면 차트를 한 번 더 확인해 볼 이유가 생깁니다.
반대로 거래가 거의 없는 코인이 검색되는 것을 막기 위해 최소 24시간 거래대금도 설정할 수 있도록 했습니다.
기본값은 10억원으로 만들어두었지만 GUI에서 자유롭게 변경할 수 있습니다.
10분마다 알아서 다시 검색
이걸 계속 보고 있을 필요도 없습니다.
기본 갱신주기는 10분입니다.
시간 갱신주기도 1분으로 줄일 수도 있고 조정이 가능합니다.
프로그램을 켜두면 업비트 KRW 마켓을 다시 스캔해서 조건에 맞는 종목을 찾아냅니다.
업비트는 공식 API를 통해 일봉 캔들 데이터를 제공하고 있기 때문에 이 데이터를 이용해 이동평균과 Envelope를 직접 계산할 수 있습니다.
그리고 검색된 티커를 더블클릭하면 해당 종목의 업비트 차트가 브라우저에서 바로 열리도록 만들었습니다. 업비트 PC 웹에서는 TradingView 차트도 제공하고 있습니다.
즉 검색기는 종목을 찾아주는 역할까지만 하고, 실제 차트 모양과 지지·저항 등은 사람이 다시 확인할 수 있습니다.
조건이 나오면 텔레그램으로 바로 알림
검색기를 만들어 놓고 하루 종일 화면을 보고 있으면 자동화의 의미가 별로 없습니다.
그래서 마지막으로 텔레그램 알림 기능을 넣었습니다.
GUI에서 자신의
BOT TOKEN
과
CHAT ID
를 입력하고 알림을 활성화하면 새로운 종목이 검색됐을 때 텔레그램으로 알려줍니다.
메시지에는 대략 다음 내용이 들어옵니다.
[업비트 ENV 신호]
종목명 / 티커
검색 조건
몇 봉 전 돌파인지
현재가
ENV 하단
하단 대비 이격률
24시간 거래대금
7일 평균 대비 거래량
업비트 차트 링크
차트 링크도 같이 보내기 때문에 휴대폰에서 알림을 받고 바로 해당 종목을 확인할 수 있습니다.
같은 종목을 10분마다 계속 보내지는 않는다
이 부분도 중요합니다.
10분마다 검색하는데 같은 종목이 계속 조건을 만족한다고 10분마다 텔레그램이 울리면 사실상 스팸이 됩니다.
그래서 새롭게 검색된 종목만 알림을 보내도록 했습니다.
예를 들어 첫 검색 결과가
A / B / C
였다면 세 종목의 알림이 옵니다.
10분 뒤에도
A / B / C
라면 아무것도 보내지 않습니다.
그런데
A / B / C / D
가 됐다면 D만 새롭게 알림이 옵니다.
B가 검색조건에서 사라졌다가 며칠 뒤 다시 조건에 진입하면 그때는 새로운 신호로 보고 다시 알림을 보냅니다.
이렇게 해두면 검색기를 계속 쳐다보고 있지 않아도 됩니다.
실제 프로그램 화면에서 변경할 수 있는 값
코드에 숫자를 박아놓지 않고 주요 설정은 GUI에서 변경하도록 만들었습니다.
ENV 기간: 기본 60일
ENV 폭: 기본 8%
최소 24시간 거래대금: 기본 10억원
자동 갱신: 기본 10분
돌파 검색: 0봉전 ~ 원하는 N봉전
Telegram: 사용 여부 / BOT TOKEN / CHAT ID
따라서 꼭 60일 / 8%를 사용할 필요도 없습니다.
각자 사용하는 매매 방식에 맞춰 숫자를 바꾸면 됩니다.
결국 핵심은 '검색기를 직접 만들 수 있다'는 것
제가 이 프로그램을 공개하는 이유도 여기에 있습니다.
이 검색기가 무조건 수익을 내는 프로그램이라서가 아닙니다.
오히려 핵심은 반대입니다.
매매하는 사람이 평소 차트에서 반복해서 찾는 조건이 있다면 그것을 직접 검색기로 만들 수 있다는 것입니다.
이번에는 아주 단순한 Envelope 하나만 사용했습니다.
하지만 여기에서
거래량 조건을 추가할 수도 있고,
장기 이동평균 방향을 넣을 수도 있고,
비트코인 추세가 좋을 때만 신호를 발생시킬 수도 있고,
주봉 조건을 하나 더 붙일 수도 있습니다.
처음부터 거대한 자동매매 프로그램을 만들 필요도 없습니다.
“조건에 맞는 종목을 찾아서 나에게 알려줘.”
여기서 시작해도 충분합니다.
그다음 실제로 사용해보면서 자신에게 필요한 조건을 하나씩 붙이면 됩니다.
전체 소스코드 공개
이번에 만든 버전은 아래 파일 그대로 공개합니다.
업비트 ENV 검색기 + 텔레그램 알림 전체 Python 코드
필요한 외부 라이브러리는 requests 정도이고 GUI는 Python 기본 tkinter를 사용했습니다.
pip install requests실행 후에는 자신의 텔레그램 Bot Token과 Chat ID를 GUI에 입력하고 테스트 전송으로 정상 작동하는지 확인하면 됩니다.
이 검색기는 매수 추천 프로그램이 아니라 조건 검색 도구입니다. Envelope 하단 돌파가 곧 상승을 보장하는 것도 아니고, 특히 알트코인은 하락 변동성이 크기 때문에 실제 운용에서는 진입 기준과 함께 어디서 틀렸다고 판단하고 손절할 것인지를 먼저 정해두는 것이 중요합니다.
# -*- coding: utf-8 -*-
"""
업비트 ENV 하단 검색기
- KRW 전 종목
- 일봉 Envelope (기본 60일, ±8%)
- 조건:
1) 현재가가 ENV 하단 아래
2) 전일봉: 시가가 ENV 하단 아래 -> 종가가 ENV 하단 위(확정 상향돌파)
3) 오늘봉: 시가가 ENV 하단 아래 -> 현재가가 ENV 하단 위(장중 상향돌파)
- 24시간 거래대금
- 오늘 거래량 / 최근 7개 완성 일봉 평균 거래량 / 거래량 비율
- 기본 10분 자동 갱신
- 결과 더블클릭 시 업비트 해당 종목 차트 열기
필요 라이브러리:
pip install requests
실행:
python 업비트_ENV60_8pct_하단검색기_20260926.py
"""
import tkinter as tk
from tkinter import ttk, messagebox
import threading
import time
import webbrowser
import requests
from datetime import datetime
API = "https://api.upbit.com/v1"
SESSION = requests.Session()
SESSION.headers.update({"User-Agent": "Mozilla/5.0 Upbit-ENV-Scanner/1.0"})
def fmt_krw(v):
try:
v = float(v)
if abs(v) >= 1_0000_0000_0000:
return f"{v/1_0000_0000_0000:.2f}조"
if abs(v) >= 1_0000_0000:
return f"{v/1_0000_0000:.1f}억"
if abs(v) >= 1_0000:
return f"{v/1_0000:.0f}만"
return f"{v:,.0f}"
except Exception:
return "-"
def fmt_price(v):
try:
v = float(v)
if v >= 1000:
return f"{v:,.0f}"
if v >= 100:
return f"{v:,.1f}"
if v >= 1:
return f"{v:,.3f}".rstrip("0").rstrip(".")
return f"{v:.8f}".rstrip("0").rstrip(".")
except Exception:
return "-"
class EnvScanner(tk.Tk):
def __init__(self):
super().__init__()
self.title("업비트 ENV 하단/돌파 검색기")
self.geometry("1450x760")
self.minsize(1200, 650)
self.running = True
self.scanning = False
self.after_id = None
self.period_var = tk.StringVar(value="60")
self.env_var = tk.StringVar(value="8")
self.min_value_var = tk.StringVar(value="10") # 억원
self.refresh_var = tk.StringVar(value="10") # 분
self.below_var = tk.BooleanVar(value=True)
self.prev_cross_var = tk.BooleanVar(value=False)
self.today_cross_var = tk.BooleanVar(value=False)
self.cross_range_var = tk.BooleanVar(value=False)
self.cross_from_var = tk.StringVar(value="0")
self.cross_to_var = tk.StringVar(value="5")
self.status_var = tk.StringVar(value="대기 중")
self.updated_var = tk.StringVar(value="-")
self.telegram_enabled_var = tk.BooleanVar(value=False)
self.telegram_token_var = tk.StringVar(value=" 본인 텔레그램 토큰 넣기 ")
self.telegram_chat_id_var = tk.StringVar(value=" 본인 텔레그램 아이디 넣기 ")
self.active_alert_markets = set()
self._build_ui()
self.protocol("WM_DELETE_WINDOW", self._close)
# 창이 뜬 직후 첫 검색
self.after(500, self.start_scan)
def _build_ui(self):
top = ttk.Frame(self, padding=10)
top.pack(fill="x")
ttk.Label(top, text="ENV 기간").grid(row=0, column=0, padx=(0, 4))
ttk.Entry(top, textvariable=self.period_var, width=7).grid(row=0, column=1, padx=(0, 12))
ttk.Label(top, text="ENV 폭(%)").grid(row=0, column=2, padx=(0, 4))
ttk.Entry(top, textvariable=self.env_var, width=7).grid(row=0, column=3, padx=(0, 12))
ttk.Label(top, text="최소 24H 거래대금(억원)").grid(row=0, column=4, padx=(0, 4))
ttk.Entry(top, textvariable=self.min_value_var, width=9).grid(row=0, column=5, padx=(0, 12))
ttk.Label(top, text="갱신(분)").grid(row=0, column=6, padx=(0, 4))
ttk.Entry(top, textvariable=self.refresh_var, width=7).grid(row=0, column=7, padx=(0, 12))
ttk.Button(top, text="지금 검색", command=self.start_scan).grid(row=0, column=8, padx=6)
filters = ttk.LabelFrame(self, text="검색 조건 (복수 선택 가능)", padding=8)
filters.pack(fill="x", padx=10, pady=(0, 8))
ttk.Checkbutton(
filters, text="현재가가 ENV 하단 아래",
variable=self.below_var
).grid(row=0, column=0, sticky="w", padx=(4, 20), pady=3)
ttk.Checkbutton(
filters, text="전일봉 확정 상향돌파",
variable=self.prev_cross_var
).grid(row=0, column=1, sticky="w", padx=(0, 20), pady=3)
ttk.Checkbutton(
filters, text="오늘 장중 상향돌파",
variable=self.today_cross_var
).grid(row=0, column=2, sticky="w", padx=(0, 20), pady=3)
ttk.Checkbutton(
filters, text="돌파 봉 범위",
variable=self.cross_range_var
).grid(row=1, column=0, sticky="w", padx=(4, 4), pady=3)
range_box = ttk.Frame(filters)
range_box.grid(row=1, column=1, columnspan=2, sticky="w", pady=3)
ttk.Entry(range_box, textvariable=self.cross_from_var, width=5).pack(side="left")
ttk.Label(range_box, text=" 봉전 ~ ").pack(side="left")
ttk.Entry(range_box, textvariable=self.cross_to_var, width=5).pack(side="left")
ttk.Label(range_box, text=" 봉전 (0=오늘 진행봉, 1=어제 완성봉, 2=2봉전...)").pack(side="left", padx=(5, 0))
tg = ttk.LabelFrame(self, text="텔레그램 신규 종목 알림", padding=7)
tg.pack(fill="x", padx=10, pady=(0, 7))
ttk.Checkbutton(tg, text="알림 사용", variable=self.telegram_enabled_var).pack(side="left", padx=(4,10))
ttk.Label(tg, text="BOT TOKEN").pack(side="left")
ttk.Entry(tg, textvariable=self.telegram_token_var, width=42, show="*").pack(side="left", padx=(4,12))
ttk.Label(tg, text="CHAT ID").pack(side="left")
ttk.Entry(tg, textvariable=self.telegram_chat_id_var, width=18).pack(side="left", padx=(4,12))
ttk.Button(tg, text="테스트 전송", command=self.test_telegram).pack(side="left")
ttk.Label(tg, text=" 같은 종목은 조건 유지 중 재알림 안 함").pack(side="left", padx=8)
info = ttk.Frame(self, padding=(10, 0, 10, 7))
info.pack(fill="x")
ttk.Label(info, textvariable=self.status_var).pack(side="left")
ttk.Label(info, text="마지막 갱신:").pack(side="right")
ttk.Label(info, textvariable=self.updated_var).pack(side="right", padx=(0, 5))
cols = (
"market", "name", "signal", "signal_bar", "price", "ma", "lower",
"gap", "trade_value", "today_vol", "avg7_vol", "vol_ratio"
)
# 중요: Treeview/Scrollbar를 처음부터 table_frame의 자식으로 생성해야
# 일부 Windows/Tk 환경에서 표 전체가 빈 화면으로 사라지는 현상이 발생하지 않는다.
table_frame = ttk.Frame(self)
table_frame.pack(fill="both", expand=True, padx=10, pady=(0, 4))
self.tree = ttk.Treeview(table_frame, columns=cols, show="headings", height=27)
headings = {
"market": "티커",
"name": "종목명",
"signal": "조건",
"signal_bar": "돌파봉",
"price": "현재가",
"ma": "MA",
"lower": "ENV 하단",
"gap": "하단 대비(%)",
"trade_value": "24H 거래대금",
"today_vol": "오늘 거래량",
"avg7_vol": "7일 평균 거래량",
"vol_ratio": "7일 평균 대비",
}
widths = {
"market": 90, "name": 120, "signal": 220, "signal_bar": 80, "price": 105,
"ma": 105, "lower": 105, "gap": 105, "trade_value": 125,
"today_vol": 125, "avg7_vol": 135, "vol_ratio": 115
}
for c in cols:
self.tree.heading(c, text=headings[c], command=lambda cc=c: self.sort_column(cc, False))
self.tree.column(c, width=widths[c], anchor="center", stretch=True)
yscroll = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview)
xscroll = ttk.Scrollbar(self, orient="horizontal", command=self.tree.xview)
self.tree.configure(yscrollcommand=yscroll.set, xscrollcommand=xscroll.set)
self.tree.pack(side="left", fill="both", expand=True)
yscroll.pack(side="right", fill="y")
xscroll.pack(fill="x", padx=10, pady=(0, 4))
self.tree.bind("<Double-1>", self.open_upbit)
bottom = ttk.Frame(self, padding=(10, 2, 10, 8))
bottom.pack(fill="x")
ttk.Label(
bottom,
text="※ 더블클릭: 업비트 차트 열기 | 돌파범위 0=오늘, 1=어제, 2=2봉전... | 7일 평균 거래량 = 최근 7개 완성 일봉"
).pack(side="left")
def read_settings(self):
try:
period = int(self.period_var.get())
env_pct = float(self.env_var.get())
min_value = float(self.min_value_var.get()) * 100_000_000
refresh_min = float(self.refresh_var.get())
if period < 2 or env_pct <= 0 or refresh_min <= 0:
raise ValueError
if not (self.below_var.get() or self.prev_cross_var.get() or self.today_cross_var.get() or self.cross_range_var.get()):
raise RuntimeError("검색 조건을 하나 이상 체크하세요.")
cross_from = int(self.cross_from_var.get())
cross_to = int(self.cross_to_var.get())
if cross_from < 0 or cross_to < 0 or cross_from > cross_to:
raise RuntimeError("돌파 봉 범위는 0 이상의 숫자로, 시작값 <= 끝값이어야 합니다.")
if cross_to > 100:
raise RuntimeError("돌파 봉 범위는 최대 100봉전까지 설정하세요.")
return period, env_pct, min_value, refresh_min, cross_from, cross_to
except RuntimeError:
raise
except Exception:
raise RuntimeError("설정값을 확인하세요. 기간/ENV폭/거래대금/갱신주기는 숫자여야 합니다.")
def api_get(self, path, params=None, retries=3):
last = None
for i in range(retries):
try:
r = SESSION.get(API + path, params=params, timeout=10)
if r.status_code == 429:
time.sleep(0.5 + i * 0.5)
continue
r.raise_for_status()
return r.json()
except Exception as e:
last = e
time.sleep(0.4 + i * 0.4)
raise last
def get_markets(self):
data = self.api_get("/market/all", {"is_details": "false"})
return [x for x in data if x["market"].startswith("KRW-")]
def get_tickers(self, market_codes):
result = {}
# ticker endpoint를 묶어서 호출하여 요청 수 절약
for i in range(0, len(market_codes), 100):
batch = market_codes[i:i+100]
data = self.api_get("/ticker", {"markets": ",".join(batch)})
for x in data:
result[x["market"]] = x
time.sleep(0.08)
return result
def get_candles(self, market, count):
return self.api_get(
"/candles/days",
{"market": market, "count": min(max(count, 10), 200)}
)
def send_telegram(self, msg, force=False):
if not force and not self.telegram_enabled_var.get():
return False
token = self.telegram_token_var.get().strip()
chat_id = self.telegram_chat_id_var.get().strip()
if not token or not chat_id:
return False
try:
r = requests.post(
f"https://api.telegram.org/bot{token}/sendMessage",
data={"chat_id": chat_id, "text": msg,
"disable_web_page_preview": True},
timeout=10
)
r.raise_for_status()
return bool(r.json().get("ok"))
except Exception as e:
print("[TELEGRAM ERROR]", e)
return False
def test_telegram(self):
if self.send_telegram("[업비트 ENV 검색기] 텔레그램 연결 테스트", force=True):
messagebox.showinfo("텔레그램", "테스트 메시지를 전송했습니다.")
else:
messagebox.showerror("텔레그램", "전송 실패. BOT TOKEN / CHAT ID를 확인하세요.")
def notify_new_rows(self, rows):
if not self.telegram_enabled_var.get():
return
if not self.telegram_token_var.get().strip() or not self.telegram_chat_id_var.get().strip():
return
current = {r["market"] for r in rows}
new_markets = current - self.active_alert_markets
for r in rows:
if r["market"] not in new_markets:
continue
ticker = r["market"].replace("KRW-", "")
msg = (
f"[업비트 ENV 신호]\n"
f"{ticker} / {r['name']}\n"
f"조건: {r['signal']}\n"
f"돌파봉: {r.get('signal_bar', '-')}\n"
f"현재가: {fmt_price(r['price'])}원\n"
f"ENV 하단: {fmt_price(r['lower'])}원\n"
f"하단 대비: {r['gap']:+.2f}%\n"
f"24H 거래대금: {fmt_krw(r['trade_value'])}원\n"
f"7일 평균 대비 거래량: {r['vol_ratio']:.1f}%\n"
f"https://upbit.com/exchange?code=CRIX.UPBIT.{r['market']}"
)
self.send_telegram(msg)
time.sleep(0.08)
# 계속 조건을 만족하는 종목은 다음 갱신 때 생략.
# 검색에서 빠졌다가 다시 들어오면 다시 신규 신호로 알림.
self.active_alert_markets = current
def start_scan(self):
if self.scanning:
return
try:
settings = self.read_settings()
except RuntimeError as e:
messagebox.showwarning("설정 확인", str(e))
return
if self.after_id:
try:
self.after_cancel(self.after_id)
except Exception:
pass
self.after_id = None
self.scanning = True
self.status_var.set("검색 시작...")
threading.Thread(target=self.scan_worker, args=(settings,), daemon=True).start()
def scan_worker(self, settings):
period, env_pct, min_value, refresh_min, cross_from, cross_to = settings
rows = []
try:
markets = self.get_markets()
codes = [m["market"] for m in markets]
names = {m["market"]: m.get("korean_name", "") for m in markets}
tickers = self.get_tickers(codes)
# 거래대금 필터를 먼저 적용하여 일봉 API 호출량 절약
candidates = []
for market in markets:
code = market["market"]
t = tickers.get(code)
if not t:
continue
trade_value = float(t.get("acc_trade_price_24h", 0) or 0)
if trade_value >= min_value:
candidates.append(code)
total = len(candidates)
need_count = min(200, period + max(12, cross_to + 3))
for idx, code in enumerate(candidates, 1):
if not self.running:
return
if idx == 1 or idx % 5 == 0:
self.after(0, lambda i=idx, n=total: self.status_var.set(f"검색 중... {i}/{n}"))
try:
candles = self.get_candles(code, need_count)
if len(candles) < period + 2:
continue
# Upbit candles: 최신 -> 과거. 계산 편의를 위해 과거 -> 최신.
cs = list(reversed(candles))
closes = [float(x["trade_price"]) for x in cs]
cur_idx = len(cs) - 1
prev_idx = len(cs) - 2
# 현재 일봉 ENV: 현재봉을 포함한 period 종가 SMA
cur_window = closes[cur_idx-period+1:cur_idx+1]
cur_ma = sum(cur_window) / period
cur_lower = cur_ma * (1.0 - env_pct / 100.0)
current_price = float(tickers[code]["trade_price"])
cur_open = float(cs[cur_idx]["opening_price"])
# 전일 확정 ENV: 전일을 끝으로 하는 period 종가 SMA
prev_window = closes[prev_idx-period+1:prev_idx+1]
prev_ma = sum(prev_window) / period
prev_lower = prev_ma * (1.0 - env_pct / 100.0)
prev_open = float(cs[prev_idx]["opening_price"])
prev_close = float(cs[prev_idx]["trade_price"])
signals = []
signal_bars = []
if self.below_var.get() and current_price < cur_lower:
signals.append("ENV 하단 아래")
if self.prev_cross_var.get() and prev_open < prev_lower and prev_close > prev_lower:
signals.append("전일 확정 상향돌파")
signal_bars.append("1봉전")
if self.today_cross_var.get() and cur_open < cur_lower and current_price > cur_lower:
signals.append("오늘 장중 상향돌파")
signal_bars.append("0봉전")
# 지정한 0~N봉전 각각에서 당시의 ENV 하단을 다시 계산해 상향돌파 여부 확인.
# 0봉전은 진행 중이므로 현재가를 종가 대신 사용한다.
if self.cross_range_var.get():
range_hits = []
for ago in range(cross_from, cross_to + 1):
bi = cur_idx - ago
if bi < period - 1:
continue
w = closes[bi-period+1:bi+1]
ma_b = sum(w) / period
lower_b = ma_b * (1.0 - env_pct / 100.0)
open_b = float(cs[bi]["opening_price"])
close_b = current_price if ago == 0 else float(cs[bi]["trade_price"])
if open_b < lower_b and close_b > lower_b:
range_hits.append(f"{ago}봉전")
if range_hits:
signals.append("범위 내 상향돌파")
signal_bars.extend(range_hits)
if not signals:
continue
# 오늘(진행 중) 거래량
today_vol = float(cs[cur_idx].get("candle_acc_trade_volume", 0) or 0)
# 최근 7개 완성 일봉 평균 거래량 (오늘봉 제외)
completed = cs[:-1]
last7 = completed[-7:]
avg7 = (
sum(float(x.get("candle_acc_trade_volume", 0) or 0) for x in last7) / len(last7)
if last7 else 0
)
vol_ratio = today_vol / avg7 * 100 if avg7 > 0 else 0
gap = (current_price / cur_lower - 1.0) * 100.0
trade_value = float(tickers[code].get("acc_trade_price_24h", 0) or 0)
rows.append({
"market": code,
"name": names.get(code, ""),
"signal": " + ".join(dict.fromkeys(signals)),
"signal_bar": ", ".join(dict.fromkeys(signal_bars)) if signal_bars else "-",
"price": current_price,
"ma": cur_ma,
"lower": cur_lower,
"gap": gap,
"trade_value": trade_value,
"today_vol": today_vol,
"avg7_vol": avg7,
"vol_ratio": vol_ratio,
})
# 공개 API에 무리 주지 않도록 약간의 간격
time.sleep(0.075)
except Exception:
# 특정 종목 실패가 전체 스캔을 중단하지 않도록 함
time.sleep(0.15)
continue
# ENV 하단 대비 이격률 낮은 순 (더 깊게 내려간 종목 우선)
rows.sort(key=lambda x: x["gap"])
self.after(0, lambda rr=rows: self.render_rows(rr))
self.notify_new_rows(rows)
except Exception as e:
self.after(0, lambda msg=str(e): messagebox.showerror("검색 오류", msg))
self.after(0, lambda: self.status_var.set("검색 오류"))
finally:
self.scanning = False
if self.running:
self.after(0, lambda m=refresh_min: self.schedule_next(m))
def render_rows(self, rows):
for item in self.tree.get_children():
self.tree.delete(item)
for r in rows:
self.tree.insert("", "end", values=(
r["market"].replace("KRW-", ""),
r["name"],
r["signal"],
r["signal_bar"],
fmt_price(r["price"]),
fmt_price(r["ma"]),
fmt_price(r["lower"]),
f'{r["gap"]:+.2f}%',
fmt_krw(r["trade_value"]),
f'{r["today_vol"]:,.0f}',
f'{r["avg7_vol"]:,.0f}',
f'{r["vol_ratio"]:.1f}%'
))
self.updated_var.set(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
self.status_var.set(f"완료: {len(rows)}개 종목")
def schedule_next(self, refresh_min):
if not self.running:
return
delay_ms = max(1, int(refresh_min * 60 * 1000))
self.after_id = self.after(delay_ms, self.start_scan)
def open_upbit(self, event=None):
sel = self.tree.selection()
if not sel:
return
values = self.tree.item(sel[0], "values")
if not values:
return
ticker = str(values[0]).upper()
# 업비트 거래소 차트 페이지
url = f"https://upbit.com/exchange?code=CRIX.UPBIT.KRW-{ticker}"
webbrowser.open_new_tab(url)
def sort_column(self, col, reverse):
items = [(self.tree.set(k, col), k) for k in self.tree.get_children("")]
def parse(v):
s = str(v).replace(",", "").replace("%", "")
mult = 1
if s.endswith("조"):
mult, s = 1e12, s[:-1]
elif s.endswith("억"):
mult, s = 1e8, s[:-1]
elif s.endswith("만"):
mult, s = 1e4, s[:-1]
try:
return (0, float(s) * mult)
except Exception:
return (1, str(v))
items.sort(key=lambda x: parse(x[0]), reverse=reverse)
for index, (_, k) in enumerate(items):
self.tree.move(k, "", index)
self.tree.heading(col, command=lambda: self.sort_column(col, not reverse))
def _close(self):
self.running = False
if self.after_id:
try:
self.after_cancel(self.after_id)
except Exception:
pass
self.destroy()
if __name__ == "__main__":
app = EnvScanner()
app.mainloop()
'자동 매매' 카테고리의 다른 글
| 업비트 서브포켓 써보니 꽤 쓸 만하다 — 투자 계좌 분리부터 API 자동매매까지 (0) | 2026.09.26 |
|---|---|
| 아비트리지 봇을 스프레드 봇으로 바꿔보기 (0) | 2026.09.15 |
| 업비트·빗썸 자동매매 시스템 트레이딩 2026년 상반기 5,400만원 수익, 하락장에서도 할 수 있다 (0) | 2026.07.15 |
| 코인 차익거래(아비트리지) 및 김치 프리미엄, 테더 단타 전략 (0) | 2026.03.03 |
| 업비트 빗썸 자동매매 1년 운영 결과 공개- 2천에서 2억원 이상으로 시드 돈복사 (0) | 2026.02.25 |
| 업비트 빗썸 바이낸스 코인 자동매매를 위한 파이썬 1분만에 설치하기 (0) | 2026.02.11 |
| 빗썸 업비트 등 코인 자동매매를 검증과 수동매매까지 가능한 다중 알트코인 매매 시스템 (0) | 2026.01.28 |
| 빗썸 신규 서비스 오픈! 업비트와 자동매매 자동차익거래 아비트리지 설정 가이드 & 주의사항 총정리 (2) | 2025.09.25 |
