Notice
Recent Posts
Recent Comments
Link
| 일 | 월 | 화 | 수 | 목 | 금 | 토 |
|---|---|---|---|---|---|---|
| 1 | 2 | |||||
| 3 | 4 | 5 | 6 | 7 | 8 | 9 |
| 10 | 11 | 12 | 13 | 14 | 15 | 16 |
| 17 | 18 | 19 | 20 | 21 | 22 | 23 |
| 24 | 25 | 26 | 27 | 28 | 29 | 30 |
| 31 |
Tags
- ETF
- 퇴직연금
- 매일매수
- OXY
- 업비트
- 직투
- 연금
- 오블완
- 토스
- 자동매매
- 삼성전자
- 티스토리챌린지
- 앱테크
- rise 200고배당커버드콜atm
- SCHD
- ace 미국30년국채액티브(h)
- 국장
- 개인연금
- Python
- JEPQ
- 강환국
- IRP
- 재테크
- chatGPT
- 삼성전자우
- S&P500
- 미국주식
- N잡
- isa
- 주식
Archives
- Today
- Total
재테크 A2Z
업비트 자동매매 봇 만들기 (1) – 텔레그램 알림 + 시그널 루프 + 로그 백테스트 본문
업비트 자동매매 시스템 구축 프로젝트의 첫 걸음으로,
텔레그램 알림 봇과 자동 시그널 루프 + 로그 저장 구조까지 구축한 과정을 기록합니다.
✅ 1. 프로젝트 구조
autobot-trader/
├── pyproject.toml
├── .env
├── backtest/
│ ├── backtest_bollinger.csv
│ ├── backtest_rsi.csv
│ ├── backtest_moving_average.csv
│ └── backtest_runner.py
├── src/
│ └── autobot_trader/
│ ├── run.py ← 자동매매 루프
│ ├── telegram_bot.py ← 텔레그램 메시지 함수
│ ├── log_signal.py ← 백테스트 및 실행 로그 기록
│ └── strategies/
│ ├── bollinger.py
│ ├── rsi.py
│ └── moving_average.py
├── pyproject.toml
├── .env
├── backtest/
│ ├── backtest_bollinger.csv
│ ├── backtest_rsi.csv
│ ├── backtest_moving_average.csv
│ └── backtest_runner.py
├── src/
│ └── autobot_trader/
│ ├── run.py ← 자동매매 루프
│ ├── telegram_bot.py ← 텔레그램 메시지 함수
│ ├── log_signal.py ← 백테스트 및 실행 로그 기록
│ └── strategies/
│ ├── bollinger.py
│ ├── rsi.py
│ └── moving_average.py
✅ 2. 텔레그램 봇 설정 (telegram_bot.py)
import os
import requests
TOKEN = os.getenv("TELEGRAM_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
def send_message(text):
url = f"https://api.telegram.org/bot{TOKEN}/sendMessage"
payload = {
"chat_id": CHAT_ID,
"text": text
}
try:
res = requests.post(url, json=payload)
res.raise_for_status()
except Exception as e:
print(f"❌ 텔레그램 전송 오류: {e}")
import requests
TOKEN = os.getenv("TELEGRAM_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
def send_message(text):
url = f"https://api.telegram.org/bot{TOKEN}/sendMessage"
payload = {
"chat_id": CHAT_ID,
"text": text
}
try:
res = requests.post(url, json=payload)
res.raise_for_status()
except Exception as e:
print(f"❌ 텔레그램 전송 오류: {e}")
.env 파일에는 다음과 같이 토큰을 등록합니다:
TELEGRAM_TOKEN=123456789:ABCDEF-telegram-token
TELEGRAM_CHAT_ID=987654321
TELEGRAM_CHAT_ID=987654321
✅ 3. 시그널 로그 기록 (log_signal.py)
import csv
import os
from datetime import datetime
def log_signal(strategy, ticker, signal, price, backtest=False):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
row = [now, ticker, strategy, signal, price]
# 일반 로그
with open("signal_log.csv", "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(row)
# 백테스트 로그
if backtest:
os.makedirs("backtest", exist_ok=True)
with open(f"backtest/backtest_{strategy}.csv", "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(row)
import os
from datetime import datetime
def log_signal(strategy, ticker, signal, price, backtest=False):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
row = [now, ticker, strategy, signal, price]
# 일반 로그
with open("signal_log.csv", "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(row)
# 백테스트 로그
if backtest:
os.makedirs("backtest", exist_ok=True)
with open(f"backtest/backtest_{strategy}.csv", "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(row)
✅ 4. 자동매매 루프 (run.py)
import time
import schedule
import pyupbit
from autobot_trader.strategies.moving_average import get_moving_average_signal
from autobot_trader.strategies.rsi import get_rsi_signal
from autobot_trader.strategies.bollinger import get_bollinger_signal
from autobot_trader.telegram_bot import send_message
from autobot_trader.log_signal import log_signal
def run_strategy(name, func):
print(f"⏱️ [{name}] 전략 실행 중...")
signal = func()
if signal in ["buy", "sell"]:
msg = f"📡 [{name}] 시그널 발생: {signal}"
send_message(msg)
price = pyupbit.get_current_price("KRW-BTC")
log_signal(name, "KRW-BTC", signal, price, backtest=True)
else:
print(f"💤 [{name}] 시그널 없음")
def main():
schedule.every(1).minutes.do(lambda: run_strategy("bollinger", get_bollinger_signal))
schedule.every(1).minutes.do(lambda: run_strategy("rsi", get_rsi_signal))
schedule.every(1).minutes.do(lambda: run_strategy("moving_average", get_moving_average_signal))
print("🚀 자동매매 루프 시작 (1분 주기)")
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == "__main__":
main()
import schedule
import pyupbit
from autobot_trader.strategies.moving_average import get_moving_average_signal
from autobot_trader.strategies.rsi import get_rsi_signal
from autobot_trader.strategies.bollinger import get_bollinger_signal
from autobot_trader.telegram_bot import send_message
from autobot_trader.log_signal import log_signal
def run_strategy(name, func):
print(f"⏱️ [{name}] 전략 실행 중...")
signal = func()
if signal in ["buy", "sell"]:
msg = f"📡 [{name}] 시그널 발생: {signal}"
send_message(msg)
price = pyupbit.get_current_price("KRW-BTC")
log_signal(name, "KRW-BTC", signal, price, backtest=True)
else:
print(f"💤 [{name}] 시그널 없음")
def main():
schedule.every(1).minutes.do(lambda: run_strategy("bollinger", get_bollinger_signal))
schedule.every(1).minutes.do(lambda: run_strategy("rsi", get_rsi_signal))
schedule.every(1).minutes.do(lambda: run_strategy("moving_average", get_moving_average_signal))
print("🚀 자동매매 루프 시작 (1분 주기)")
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == "__main__":
main()
✅ 5. 실행 방법
Poetry 환경에서 실행
poetry run python src/autobot_trader/run.py
1분마다 시그널을 체크하며, 시그널 발생 시:
- 텔레그램으로 알림 전송
- 로그 및 백테스트 로그 자동 저장
📌 마무리
이제 텔레그램 봇과 함께 시그널을 실시간으로 추적하고,
추후 분석을 위한 백테스트 CSV까지 자동 저장되는 자동매매 기반 구조가 완성되었습니다.
📈 백테스트 분석 및 수익률 시각화 + HTML 리포트 자동 생성
'코딩 & 파이썬' 카테고리의 다른 글
| 업비트 자동매매 봇 만들기 (3) 업비트 OpenAPI 연동 및 IP 주소 등록 방법 (0) | 2025.05.24 |
|---|---|
| 업비트 자동매매 봇 만들기 (2) 백테스트 분석 및 수익률 시각화 + HTML 리포트 자동 생성 (0) | 2025.05.24 |
| 자동매매서버구축 X200 vs. N100 (0) | 2025.05.20 |
| 2025.05.19 VS Code 가상환경 연동 ~ 전략모듈 실행테스트 (0) | 2025.05.19 |
| Poetry 설치 후 GitHub 연동 시 발생한 에러와 해결 방법 (Windows 기준) (0) | 2025.05.19 |
