์ผ์ฑ์ ๊ธฐ ์ฃผ๊ฐ ์์๋ณด๊ธฐ
์์ต์ด ๋์ ๋ ํผ๋ฐ์ค๋ฅผ ์ ์ ํ๊ธฐ ์ํด OKX์ MEXC๋ฅผ 1์๊ถ์ผ๋ก ์ค์ ํฉ๋๋ค. ๋ ๊ฑฐ๋์ ๋ชจ๋ 1,400๋ง ๊ฐ ์ด์ ๋ด ์์ฑ ์ค์ ๊ณผ ๋์ ์ฌ์ฉ์ ํ๊ฐ๋ฅผ ๋ณด์ ํ๊ณ ์์ต๋๋ค.
์๋ 5๊ฐ ์นดํ ๊ณ ๋ฆฌ๋ก ์ง๋ฌธ์ ๋๋ ๋ ํผ๋ฐ์ค๋ฅผ ์ฒด๊ณ์ ์ผ๋ก ๋ถ์ํฉ๋๋ค. ๊ฐ ์ง๋ฌธ์ ๊ตฌ์ฒด์ ํค์๋(์ ๋ต·API·๋ฐฑํ ์คํธ ๊ฒฐ๊ณผ)๋ก ๊ฒ์ํ๋ฉฐ, Python ์ฝ๋ ๊ตฌํ์ ์ด์ ์ ๋ง์ถฅ๋๋ค.
๋ ํผ๋ฐ์ค์ ๋งค์/๋งค๋ ๊ท์น, RSI/MACD ์งํ ๊ฒฐํฉ ์กฐ๊ฑด ์ถ์ถ
์ ๋นํธ·๋ฐ์ด๋ธ์ค·Bybit ccxt ๋ผ์ด๋ธ๋ฌ๋ฆฌ๋ก 1์ด ์ฃผ๋ฌธ ์คํ ์ฝ๋
2025๋ BTC/USDT ๊ณผ๊ฑฐ ๋ฐ์ดํฐ๋ก ์์ต ์ฌํ์ฑ ๊ฒ์ฆ
ํฌ์ง์ ์ฌ์ด์ง, ํธ๋ ์ผ๋ง ์คํ, ๋ฌ๊ทธํ ๊ฐ์ง AI ํํฐ
Docker + Telegram ์๋ฆผ์ผ๋ก 24/7 ํด๋ผ์ฐ๋ ์๋ ์ด์
# ── ์ฝ์ธ ์๋๋งค๋งค ํต์ฌ ์ ๋ต (RSI + MACD ๊ฒฐํฉ) ────────────────── import ccxt import pandas as pd import ta class TradingStrategy: def __init__(self, exchange_id='okx', symbol='BTC/USDT'): self.exchange = getattr(ccxt, exchange_id)({ 'apiKey': 'YOUR_API_KEY', 'secret': 'YOUR_SECRET', 'enableRateLimit': True, }) self.symbol = symbol self.rsi_buy = 30 # ๊ณผ๋งค๋ → ๋งค์ ์ ํธ self.rsi_sell = 70 # ๊ณผ๋งค์ → ๋งค๋ ์ ํธ self.risk_pct = 0.01 # ๋ฆฌ์คํฌ 1% ๋ฃฐ def fetch_ohlcv(self, timeframe='1h', limit=200): """์บ๋ค ๋ฐ์ดํฐ ๊ฐ์ ธ์ค๊ธฐ""" data = self.exchange.fetch_ohlcv(self.symbol, timeframe, limit=limit) df = pd.DataFrame(data, columns=['timestamp','open','high','low','close','volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df def calc_indicators(self, df): """RSI + MACD ์งํ ๊ณ์ฐ""" df['rsi'] = ta.momentum.RSIIndicator(df['close'], window=14).rsi() macd = ta.trend.MACD(df['close']) df['macd'] = macd.macd() df['macd_signal'] = macd.macd_signal() df['ema200'] = ta.trend.EMAIndicator(df['close'], window=200).ema_indicator() return df def generate_signal(self, df): """๋งค์/๋งค๋ ์ ํธ ์์ฑ""" latest = df.iloc[-1] prev = df.iloc[-2] # ── ๋งค์ ์กฐ๊ฑด: RSI ๊ณผ๋งค๋ + MACD ๊ณจ๋ ํฌ๋ก์ค + EMA200 ์ buy_signal = ( latest['rsi'] < self.rsi_buy and latest['macd'] > latest['macd_signal'] and prev['macd'] < prev['macd_signal'] and latest['close'] > latest['ema200'] ) # ── ๋งค๋ ์กฐ๊ฑด: RSI ๊ณผ๋งค์ + MACD ๋ฐ๋ํฌ๋ก์ค sell_signal = ( latest['rsi'] > self.rsi_sell and latest['macd'] < latest['macd_signal'] and prev['macd'] > prev['macd_signal'] ) if buy_signal: return 'BUY' if sell_signal: return 'SELL' return 'HOLD'
ํ๊ตญ ๊ฑฐ๋์(์
๋นํธ)์ ๊ธ๋ก๋ฒ(๋ฐ์ด๋ธ์ค/Bybit)์ ์ง์ํ๋ API๋ฅผ ccxt ๋ผ์ด๋ธ๋ฌ๋ฆฌ๋ก ๋ฒค์น๋งํนํฉ๋๋ค.
# ── ๋ฉํฐ ๊ฑฐ๋์ API ์ฐ๋ (ccxt) ──────────────────────────────── import ccxt, time, logging class OrderExecutor: def __init__(self, exchange_id: str, config: dict): self.ex = getattr(ccxt, exchange_id)(config) self.ex.load_markets() logging.basicConfig(level=logging.INFO) def get_balance(self, currency='USDT') -> float: bal = self.ex.fetch_balance() return bal['free'].get(currency, 0) def place_market_order(self, symbol, side, amount): """์์ฅ๊ฐ ์ฃผ๋ฌธ (retry ํฌํจ)""" for attempt in range(3): try: order = self.ex.create_market_order(symbol, side, amount) logging.info(f"✅ {side} ์ฃผ๋ฌธ ์๋ฃ: {order['id']}") return order except ccxt.NetworkError as e: logging.warning(f"๋คํธ์ํฌ ์ค๋ฅ (์ฌ์๋ {attempt+1}/3): {e}") time.sleep(2 ** attempt) except ccxt.ExchangeError as e: logging.error(f"๊ฑฐ๋์ ์ค๋ฅ: {e}") raise def place_limit_with_stoploss(self, symbol, side, amount, price, sl_pct=0.02): """๋ฆฌ๋ฐ ์ฃผ๋ฌธ + ์คํ๋ก์ค ๋์ ์ค์ """ order = self.ex.create_limit_order(symbol, side, amount, price) sl_price = price * (1 - sl_pct) if side == 'buy' else price * (1 + sl_pct) # OKX ์คํ๋ก์ค (๊ฑฐ๋์๋ง๋ค ํ๋ผ๋ฏธํฐ ์์ด) sl_params = {'stopLoss': {'triggerPrice': sl_price, 'type': 'market'}} logging.info(f"๐ก️ ์คํ๋ก์ค ์ค์ : {sl_price:.2f}") return order # ── ์ฌ์ฉ ์์ ── executor = OrderExecutor('okx', { 'apiKey': 'YOUR_API_KEY', 'secret': 'YOUR_SECRET', 'password': 'YOUR_PASSPHRASE', # OKX ํ์ 'sandbox': True, # ํ ์คํธ๋ท ๋จผ์ ! }) bal = executor.get_balance() print(f"๐ฐ USDT ์๊ณ : {bal}")
upbit, ๋ฐ์ด๋ธ์ค๋ binance, Bybit์ bybit์ผ๋ก exchange_id๋ฅผ ๋ฐ๊พธ๋ฉด ๋์ผํ๊ฒ ์๋ํฉ๋๋ค. ๋ฐ๋์ sandbox: True๋ก ํ
์คํธ ๋จผ์ !๊ณผ๊ฑฐ ๋ฐ์ดํฐ๋ก ์์ต ์ฌํ์ฑ์ ํ์ธํฉ๋๋ค. ์คํ ๋น์จ, ์ต๋ ๋๋ก๋ค์ด, ์ฐํ๊ท ์์ต๋ฅ ๊ณ์ฐ์ด ํต์ฌ์ ๋๋ค.
# ── ๋ฐฑํ ์คํ ์์ง (์คํ๋น์จ·MDD·์์ต๋ฅ ) ──────────────────────── import numpy as np import pandas as pd class Backtester: def __init__(self, initial_capital=10000, fee_rate=0.001): self.capital = initial_capital self.fee_rate = fee_rate # ์์๋ฃ 0.1% self.trades = [] self.equity = [initial_capital] def run(self, df: pd.DataFrame, signals: list): """์๊ทธ๋ ๊ธฐ๋ฐ ๋ฐฑํ ์คํธ ์คํ""" position, entry = 0, 0 for i, row in df.iterrows(): sig = signals[i] price = row['close'] if sig == 'BUY' and position == 0: qty = (self.capital * 0.95) / price position, entry = qty, price self.capital -= qty * price * (1 + self.fee_rate) elif sig == 'SELL' and position > 0: pnl = position * (price - entry) * (1 - self.fee_rate) self.capital += position * price * (1 - self.fee_rate) self.trades.append({'entry': entry, 'exit': price, 'pnl': pnl}) position = 0 self.equity.append(self.capital + position * price) def calc_metrics(self): """์คํ๋น์จ · ์ต๋๋๋ก๋ค์ด · ์น๋ฅ ๊ณ์ฐ""" eq = np.array(self.equity) rets = np.diff(eq) / eq[:-1] sharpe = rets.mean() / (rets.std() + 1e-9) * np.sqrt(252 * 24) peak = np.maximum.accumulate(eq) mdd = ((eq - peak) / peak).min() pnls = [t['pnl'] for t in self.trades] win_rate = sum(1 for p in pnls if p > 0) / len(pnls) if pnls else 0 total_return = (eq[-1] - eq[0]) / eq[0] * 100 return { '๐ ์ด ์์ต๋ฅ ' : f'{total_return:.2f}%', '⚡ ์คํ ๋น์จ' : f'{sharpe:.3f}', '๐ ์ต๋ ๋๋ก๋ค์ด': f'{mdd*100:.2f}%', '๐ฏ ์น๋ฅ ' : f'{win_rate*100:.1f}%', '๐ข ์ด ๊ฑฐ๋ ํ์': len(self.trades), }
๊ณ ์์ต ๋ด์ ์์ค ๋ฐฉ์ง ๋ก์ง์ ๋ณต์ ํฉ๋๋ค. ํฌ์ง์ ์ฌ์ด์ง๋ถํฐ ๋ฌ๊ทธํ ๊ฐ์ง๊น์ง ์๋ฒฝ ๊ตฌํํฉ๋๋ค.
ํ ๋ฒ์ ๊ฑฐ๋์์ ์ ์ฒด ์๋ณธ์ ์ต๋ 1~2%๋ง ๋ฆฌ์คํฌ๋ก ์ค์ ํ์ธ์. ๋ ๋ฒ๋ฆฌ์ง ์ฌ์ฉ ์ ์ฒญ์ฐ ๋ผ์ธ์ ๋ฐ๋์ ๊ณ์ฐํ๊ณ , ํ ์คํธ๋ท์์ ์ต์ 1์ฃผ์ผ ๊ฒ์ฆ ํ ๋ฉ์ธ๋ท ์ ํํ์ธ์.
# ── ๋ฆฌ์คํฌ ๊ด๋ฆฌ ๋ชจ๋ ──────────────────────────────────────────── class RiskManager: def __init__(self, total_capital: float, risk_per_trade: float = 0.01, max_daily_loss: float = 0.05): self.capital = total_capital self.risk_per_trade = risk_per_trade # 1% ๋ฆฌ์คํฌ ๋ฃฐ self.max_daily_loss = max_daily_loss # ํ๋ฃจ ์ต๋ ์์ค 5% self.daily_loss = 0.0 self.trailing_high = 0.0 def calc_position_size(self, entry: float, stop_loss: float) -> float: """ํฌ์ง์ ์ฌ์ด์ง: ๋ฆฌ์คํฌ ๊ธ์ก ÷ ์์คํญ""" risk_amount = self.capital * self.risk_per_trade loss_per_unit = abs(entry - stop_loss) if loss_per_unit == 0: return 0 qty = risk_amount / loss_per_unit return round(qty, 6) def update_trailing_stop(self, current_price: float, trail_pct: float = 0.03) -> float: """ํธ๋ ์ผ๋ง ์คํ ์ ๋ฐ์ดํธ (3% ์ถ์ )""" self.trailing_high = max(self.trailing_high, current_price) stop_price = self.trailing_high * (1 - trail_pct) return stop_price def check_volatility_filter(self, df) -> bool: """๋ณ๋์ฑ ๊ธ๋ฑ ์ ๊ฑฐ๋ ์ฐจ๋จ (ATR ๊ธฐ๋ฐ)""" import ta atr = ta.volatility.AverageTrueRange( df['high'], df['low'], df['close'], window=14 ).average_true_range().iloc[-1] avg_atr = ta.volatility.AverageTrueRange( df['high'], df['low'], df['close'], window=14 ).average_true_range().mean() if atr > avg_atr * 2.5: print("⚠️ ๋ณ๋์ฑ ๊ธ๋ฑ ๊ฐ์ง! ๊ฑฐ๋ ์ฐจ๋จ") return False # ๊ฑฐ๋ ๊ธ์ง return True def can_trade_today(self, current_loss: float) -> bool: """์ผ์ผ ์ต๋ ์์ค ์ด๊ณผ ์ ๊ฑฐ๋ ์ค๋จ""" self.daily_loss += current_loss if self.daily_loss < -(self.capital * self.max_daily_loss): print("๐ซ ์ผ์ผ ์ต๋ ์์ค ๋๋ฌ → ๊ฑฐ๋ ์ค๋จ") return False return True
ํด๋ผ์ฐ๋ ์๋ฒ์์ 24/7 ๊ฐ๋๋๋๋ก Docker + Telegram ์๋ฆผ์ ์ค์ ํฉ๋๋ค.
# Dockerfile ───────────────────────────────────────────────── FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "-u", "main.py"] # docker-compose.yml ────────────────────────────────────────── # version: '3.8' # services: # trading-bot: # build: . # restart: always # ํฌ๋์ ์ ์๋ ์ฌ์์ # env_file: .env # volumes: # - ./logs:/app/logs # requirements.txt ๋ด์ฉ # ccxt==4.3.0 # pandas==2.2.0 # ta==0.11.0 # python-telegram-bot==21.0 # flask==3.0.0 # python-dotenv==1.0.0
# ── Telegram ์๋ฆผ ๋ชจ๋ ────────────────────────────────────────── import requests, datetime class TelegramNotifier: def __init__(self, token: str, chat_id: str): self.url = f"https://api.telegram.org/bot{token}/sendMessage" self.chat_id = chat_id def send(self, msg: str): requests.post(self.url, json={ 'chat_id': self.chat_id, 'text': msg, 'parse_mode': 'HTML' }) def notify_trade(self, side, symbol, qty, price, pnl=None): emoji = '๐ข' if side=='BUY' else '๐ด' pnl_str = f"\n๐ฐ PnL: {pnl:+.2f} USDT" if pnl else "" msg = ( f"{emoji} <b>{side}</b> ์ฒด๊ฒฐ\n" f"๐ ์ฌ๋ณผ: {symbol}\n" f"๐ฆ ์๋: {qty}\n" f"๐ต ๊ฐ๊ฒฉ: {price:,.2f}\n" f"⏰ {datetime.datetime.now():%Y-%m-%d %H:%M:%S}" + pnl_str ) self.send(msg) def notify_error(self, error: str): self.send(f"๐จ <b>์๋ฌ ๋ฐ์</b>\n{error}")
# ── TradingView ์นํ → ์๋ ์ฃผ๋ฌธ ──────────────────────────────── from flask import Flask, request, jsonify import hmac, hashlib, os app = Flask(__name__) WEBHOOK_SECRET = os.getenv('WEBHOOK_SECRET', 'my_secret_key') @app.route('/webhook', methods=['POST']) def webhook(): # ์ํฌ๋ฆฟ ๊ฒ์ฆ sig = request.headers.get('X-Signature', '') body = request.get_data() expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest() if not hmac.compare_digest(sig, expected): return jsonify({'error': 'unauthorized'}), 401 data = request.json action = data.get('action') # 'buy' or 'sell' symbol = data.get('symbol', 'BTC/USDT') amount = float(data.get('amount', 0.001)) # ์ฃผ๋ฌธ ์คํ (executor๋ ์ ์ญ ์ธ์คํด์ค) if action in ('buy', 'sell'): result = executor.place_market_order(symbol, action, amount) notifier.notify_trade(action.upper(), symbol, amount, result['price']) return jsonify({'status': 'ok', 'order_id': result['id']}) return jsonify({'status': 'ignored'}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)
Python 3.10+, ccxt, pandas, ta-lib ์ค์น. GitHub์์ MEXC/OKX ์คํ์์ค ํด๋ก .
๊ธฐ๋ณธ ๋ด ๊ตฌํ (์บ๋ค ๋ถ์ → RSI>70 ๋งค๋ ๋ก์ง), ๋ก์ปฌ ํ ์คํธ. ์ ์ฝ๋ ๋ณต๋ถ ํ API ํค ์ ๋ ฅ.
2025๋ ๋ฐ์ดํฐ๋ก ์์ต 20%+ ๋ชฉํ. ํ์ดํผํ๋ผ๋ฏธํฐ ์ต์ ํ (RSI ๊ธฐ๊ฐ, MACD ํ๋ผ๋ฏธํฐ ๋ฑ).
ํ ์คํธ๋ท์์ 1์ฃผ ์คํ ํ ๋ฉ์ธ๋ท ์ ํ. 0.1 BTC ๋๋ ์์ก(~$100)์ผ๋ก ์์.
GPT-4 ํตํฉ์ผ๋ก ๋์ ์ ๋ต ์์ฑ. ์์ฅ ์ํฉ๋ณ ํ๋ผ๋ฏธํฐ ์๋ ์กฐ์ ๊ตฌํ.
# ── ํ๊ฒฝ ์ค์น (Ubuntu 22.04 / macOS / Windows WSL2) ───────────── # 1. ๊ฐ์ํ๊ฒฝ ์์ฑ python3 -m venv trading_venv source trading_venv/bin/activate # Windows: trading_venv\Scripts\activate # 2. ํต์ฌ ๋ผ์ด๋ธ๋ฌ๋ฆฌ ์ค์น pip install ccxt pandas ta numpy requests flask python-dotenv pip install python-telegram-bot aiohttp websockets # 3. TA-Lib ์ค์น (์์คํ ๋ผ์ด๋ธ๋ฌ๋ฆฌ ํ์) # Ubuntu: sudo apt-get install -y ta-lib && pip install TA-Lib # macOS: brew install ta-lib && pip install TA-Lib # 4. MEXC/OKX ์คํ์์ค ํด๋ก (๋ ํผ๋ฐ์ค ๋ถ์) git clone https://github.com/mexc-api/mexc-python-sdk git clone https://github.com/okxapi/python-okx # 5. .env ํ์ผ ์์ฑ (API ํค ๋ณด์ ๊ด๋ฆฌ) cat > .env << EOF OKX_API_KEY=your_api_key_here OKX_SECRET=your_secret_here OKX_PASSPHRASE=your_passphrase_here TELEGRAM_TOKEN=your_bot_token_here TELEGRAM_CHAT_ID=your_chat_id_here EOF
์ค์ ์์ต์ ์์ฅ ๋ณ๋์ฑ์ ๋ฌ๋ ค ์์ผ๋ฉฐ ์ด๋ค ๋ด๋ ์์ต์ ๋ณด์ฅํ์ง ์์ต๋๋ค. ๋ฐ๋์ ๋ฆฌ์คํฌ 1% ๋ฃฐ์ ์ค์ํ๊ณ , ์์ด๋ ๋๋ ๊ธ์ก๋ง ํฌ์ํ์ธ์. ์ด ๊ฐ์ด๋๋ ๊ต์ก ๋ชฉ์ ์ด๋ฉฐ ํฌ์ ๊ถ์ ๊ฐ ์๋๋๋ค.
์ฝ์ธ ๋งค๋งค ์๋ํ ๊ฐ์ด๋ · OKX & MEXC ๋ฒค์น๋งํน
๋ณธ ์๋ฃ๋ ๊ต์ก ๋ชฉ์ ์ผ๋ก ์ ์๋์์ต๋๋ค. ํฌ์๋ ๋ณธ์ธ ์ฑ ์์ด๋ฉฐ ์์ฅ ๋ณ๋์ฑ์ผ๋ก ์ธํด ์์ค์ด ๋ฐ์ํ ์ ์์ต๋๋ค.
์ฌ๋ฌ๋ถ์ ํ์์ด ํฐ ํ์ด ๋ฉ๋๋ค!
ํ์ํ๋ฌ ๊ฐ๊ธฐ
๋๊ธ
๋๊ธ ์ฐ๊ธฐ