import os import logging import asyncio from collections import Counter import pyupbit import pandas as pd import numpy as np import aiohttp from datetime import datetime, timedelta from persistence import save_session, save_position, delete_position from chart_renderer import render_chart logger = logging.getLogger(__name__) ENTRY_SCORE_THRESHOLD = 11 MIN_SIGNAL_BARS = 80 MIN_ATR_PCT = 0.8 MAX_ATR_PCT = 4.8 MAX_DAILY_ATR_PCT = 3.8 DAILY_TREND_BARS = 60 DAILY_FETCH_COUNT = 70 DAILY_SCREEN_CACHE_SECONDS = 3600 DAILY_SCREEN_BATCH_SIZE = 5 DAILY_SCREEN_BATCH_DELAY_SECONDS = 0.65 ORDER_FLOW_TRADE_COUNT = 200 ORDER_FLOW_BOOK_LEVELS = 5 ORDER_FLOW_MIN_TRADES = 30 ORDER_FLOW_MAX_SPREAD_BPS = 30.0 MAJOR_TICKERS = { "KRW-BTC", "KRW-ETH", "KRW-XRP", "KRW-SOL", "KRW-ADA", } STRATEGY_PROFILES = { "major": { "label": "메이저 추세", "entry_threshold": 11, "min_atr_pct": 0.2, "max_atr_pct": 2.8, "min_daily_atr_pct": 0.35, "max_daily_atr_pct": 3.4, "cooldown_win": 4, "cooldown_loss": 8, }, "alt": { "label": "알트 돌파", "entry_threshold": 10, "min_atr_pct": 1.0, "max_atr_pct": 6.5, "min_daily_atr_pct": 0.8, "max_daily_atr_pct": 8.0, "cooldown_win": 6, "cooldown_loss": 10, }, } TRADE_TIME_FORMAT = "%Y-%m-%d %H:%M:%S" LEGACY_TRADE_TIME_FORMAT = "%m-%d %H:%M" def reconstruct_virtual_account( session: dict, positions: list[dict], fee_rate: float = 0.0005, ) -> tuple[float, dict[str, float]]: """Restore paper cash and coin balances from persisted session state.""" virtual_coins: dict[str, float] = {} invested_cash = 0.0 for pos in positions: if pos.get("status") not in ("holding", "buying"): continue coin = pos.get("coin") volume = float(pos.get("buy_volume") or 0) buy_price = float(pos.get("buy_price") or 0) if coin and volume > 0: virtual_coins[coin] = virtual_coins.get(coin, 0.0) + volume if buy_price > 0 and volume > 0: invested_cash += buy_price * volume / (1 - fee_rate) saved_cash = session.get("virtual_krw") if saved_cash is None: saved_cash = ( float(session.get("total_budget") or 0) + float(session.get("total_realized") or 0) - invested_cash ) return max(float(saved_cash), 0.0), virtual_coins # ============================================================ # Technical Indicators # ============================================================ def calc_rsi(close: pd.Series, period: int = 14) -> pd.Series: delta = close.diff() gain = delta.where(delta > 0, 0).rolling(period).mean() loss = (-delta.where(delta < 0, 0)).rolling(period).mean() rs = gain / loss return 100 - (100 / (1 + rs)) def calc_macd(close: pd.Series, fast=12, slow=26, signal=9): ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line return macd_line, signal_line, histogram def calc_bollinger(close: pd.Series, period=20, num_std=2): sma = close.rolling(period).mean() std = close.rolling(period).std() return sma + num_std * std, sma, sma - num_std * std def calc_ema(close: pd.Series, period: int) -> pd.Series: return close.ewm(span=period, adjust=False).mean() def calc_stochastic(df: pd.DataFrame, k_period=14, d_period=3): low_min = df["low"].rolling(k_period).min() high_max = df["high"].rolling(k_period).max() denom = high_max - low_min k = ((df["close"] - low_min) / denom.replace(0, np.nan)) * 100 d = k.rolling(d_period).mean() return k, d def calc_atr(df: pd.DataFrame, period=14) -> pd.Series: high = df["high"] low = df["low"] close = df["close"] tr = pd.concat([ high - low, (high - close.shift()).abs(), (low - close.shift()).abs(), ], axis=1).max(axis=1) return tr.rolling(period).mean() def calc_atr_pct(df: pd.DataFrame, period: int = 14) -> float | None: if df is None or len(df) < period: return None atr = calc_atr(df, period) cur = df["close"].iloc[-1] atr_value = atr.iloc[-1] if pd.isna(atr_value) or not cur: return None return float((atr_value / cur) * 100) def rank_market_tickers( tickers: list[str], volume_map: dict[str, float], ) -> list[str]: """전체 종목을 유지한 채 거래대금이 큰 순서로 처리한다.""" return sorted(tickers, key=lambda ticker: volume_map.get(ticker, 0), reverse=True) def analyze_order_flow( orderbook: dict | None, trades: list[dict] | None, ) -> dict | None: """호가와 실제 체결을 결합해 단기 매수 수급을 점수화한다.""" if not orderbook or not trades or len(trades) < ORDER_FLOW_MIN_TRADES: return None units = orderbook.get("orderbook_units") or [] if not units: return None top_units = units[:ORDER_FLOW_BOOK_LEVELS] try: bid_notional = sum( float(unit["bid_price"]) * float(unit["bid_size"]) for unit in top_units ) ask_notional = sum( float(unit["ask_price"]) * float(unit["ask_size"]) for unit in top_units ) best_bid = float(top_units[0]["bid_price"]) best_ask = float(top_units[0]["ask_price"]) except (KeyError, TypeError, ValueError): return None book_total = bid_notional + ask_notional midpoint = (best_bid + best_ask) / 2 if book_total <= 0 or midpoint <= 0 or best_ask < best_bid: return None book_buy_ratio = bid_notional / book_total spread_bps = ((best_ask - best_bid) / midpoint) * 10_000 signed_trades = [] try: for trade in trades: side = trade.get("ask_bid") if side not in ("BID", "ASK"): continue notional = float(trade["trade_price"]) * float(trade["trade_volume"]) if notional > 0: signed_trades.append((side, notional)) except (KeyError, TypeError, ValueError): return None if len(signed_trades) < ORDER_FLOW_MIN_TRADES: return None notionals = np.array([notional for _, notional in signed_trades], dtype=float) total_notional = float(notionals.sum()) if total_notional <= 0: return None buy_notional = sum( notional for side, notional in signed_trades if side == "BID" ) trade_buy_ratio = buy_notional / total_notional large_cutoff = float(np.percentile(notionals, 90)) large_trades = [ (side, notional) for side, notional in signed_trades if notional >= large_cutoff ] large_total = sum(notional for _, notional in large_trades) if large_total <= 0: return None large_buy_ratio = ( sum(notional for side, notional in large_trades if side == "BID") / large_total ) score = 0 reasons = [] negative_signals = 0 if book_buy_ratio >= 0.57: score += 1 reasons.append(f"호가매수우세({book_buy_ratio:.0%})") elif book_buy_ratio <= 0.43: score -= 1 negative_signals += 1 reasons.append(f"호가매도우세({1 - book_buy_ratio:.0%})") if trade_buy_ratio >= 0.55: score += 1 reasons.append(f"체결매수우세({trade_buy_ratio:.0%})") elif trade_buy_ratio <= 0.45: score -= 1 negative_signals += 1 reasons.append(f"체결매도우세({1 - trade_buy_ratio:.0%})") if large_buy_ratio >= 0.58: score += 1 reasons.append(f"대량매수우세({large_buy_ratio:.0%})") elif large_buy_ratio <= 0.42: score -= 1 negative_signals += 1 reasons.append(f"대량매도우세({1 - large_buy_ratio:.0%})") spread_blocked = spread_bps > ORDER_FLOW_MAX_SPREAD_BPS if spread_blocked: reasons.append(f"스프레드과다({spread_bps:.1f}bp)") return { "score": score, "blocked": spread_blocked or negative_signals >= 2, "book_buy_ratio": round(book_buy_ratio, 4), "trade_buy_ratio": round(trade_buy_ratio, 4), "large_buy_ratio": round(large_buy_ratio, 4), "spread_bps": round(spread_bps, 2), "trade_count": len(signed_trades), "reasons": reasons, } def get_completed_candle_df( df: pd.DataFrame | None, interval_minutes: int, reference_time: datetime | None = None, ) -> pd.DataFrame | None: """Upbit 응답의 마지막 행이 진행 중 캔들이면 제외한다.""" if df is None or len(df) < 2: return df ref = reference_time or datetime.now() if hasattr(ref, "to_pydatetime"): ref = ref.to_pydatetime() last_ts = df.index[-1].to_pydatetime() cutoff = last_ts + timedelta(minutes=interval_minutes) if ref < cutoff: return df.iloc[:-1].copy() return df.copy() def get_latest_closed_candle_start( interval_minutes: int, reference_time: datetime | None = None, ) -> datetime: now = reference_time or datetime.now() minute_bucket = (now.minute // interval_minutes) * interval_minutes current_candle_start = now.replace( minute=minute_bucket, second=0, microsecond=0 ) return current_candle_start - timedelta(minutes=interval_minutes) def format_trade_time(value: datetime | None = None) -> str: return (value or datetime.now()).strftime(TRADE_TIME_FORMAT) def parse_trade_time(value: str | None) -> datetime | None: if not value: return None try: return datetime.strptime(value, TRADE_TIME_FORMAT) except ValueError: pass try: parsed = datetime.strptime(value, LEGACY_TRADE_TIME_FORMAT) return parsed.replace(year=datetime.now().year) except ValueError: return None # ============================================================ # Trend Filter - 일봉 기반 대세 판단 # ============================================================ def check_trend(df_daily: pd.DataFrame) -> tuple[str, int]: """일봉 EMA로 대세 판단. Returns (trend, score_bonus). 하락 추세는 상대강도 전략에서 사용할 수 있도록 -2를 반환한다. """ if df_daily is None or len(df_daily) < 60: return "neutral", 0 close = df_daily["close"] ema20 = calc_ema(close, 20) ema50 = calc_ema(close, 50) cur = close.iloc[-1] above_ema20 = cur > ema20.iloc[-1] above_ema50 = cur > ema50.iloc[-1] ema20_rising = ema20.iloc[-1] > ema20.iloc[-3] if not above_ema20 and not above_ema50 and not ema20_rising: return "down", -2 if above_ema20 and above_ema50 and ema20_rising: return "up", 2 return "neutral", 0 def get_entry_trend_adjustment( ticker: str, asset_trend: str, btc_trend: str, ) -> int | None: """메이저는 강세장만, 알트는 BTC 위험 점수를 반영해 선별한다.""" if asset_trend != "up": return None adjustment = 2 if ticker == "KRW-BTC": return adjustment if get_strategy_kind(ticker) == "major" and btc_trend != "up": return None adjustment += {"up": 2, "neutral": 0, "down": -2}.get(btc_trend, 0) return adjustment # ============================================================ # Coin Analysis - 메이저 추세 / 알트 돌파 분리 # ============================================================ def get_strategy_kind(ticker: str | None) -> str: if ticker in MAJOR_TICKERS: return "major" return "alt" def get_strategy_profile(ticker_or_kind: str | None) -> dict: kind = ticker_or_kind if ticker_or_kind in STRATEGY_PROFILES else get_strategy_kind(ticker_or_kind) return STRATEGY_PROFILES[kind] def _build_signal_result( strategy_kind: str, price: float, score: int, reasons: list[str], target_profit: float, stop_loss: float, atr_pct: float, rsi: float, vol_ratio: float, ) -> dict: profile = get_strategy_profile(strategy_kind) return { "price": price, "score": score, "reasons": reasons, "target_profit": round(target_profit, 2), "stop_loss": round(stop_loss, 2), "atr_pct": round(atr_pct, 2), "rsi": rsi, "vol_ratio": vol_ratio, "strategy_kind": strategy_kind, "strategy_name": profile["label"], } def analyze_major_coin(df: pd.DataFrame) -> dict | None: if df is None or len(df) < MIN_SIGNAL_BARS: return None close = df["close"] volume = df["volume"] cur = close.iloc[-1] profile = STRATEGY_PROFILES["major"] rsi = calc_rsi(close) macd_line, signal_line, macd_hist = calc_macd(close) ema5 = calc_ema(close, 5) ema20 = calc_ema(close, 20) ema60 = calc_ema(close, 60) atr = calc_atr(df) vol_avg = volume.rolling(20).mean() cur_rsi = rsi.iloc[-1] cur_hist = macd_hist.iloc[-1] prev_hist = macd_hist.iloc[-2] cur_atr = atr.iloc[-1] cur_vol_avg = vol_avg.iloc[-1] cur_vol_ratio = (volume.iloc[-1] / cur_vol_avg) if cur_vol_avg > 0 else 0 if ( pd.isna(cur_rsi) or pd.isna(cur_hist) or pd.isna(prev_hist) or pd.isna(cur_atr) or cur_atr <= 0 or pd.isna(cur_vol_avg) ): return None atr_pct = (cur_atr / cur) * 100 if atr_pct < profile["min_atr_pct"] or atr_pct > profile["max_atr_pct"]: return None if cur_rsi < 42 or cur_rsi > 66: return None if ema20.iloc[-1] <= ema60.iloc[-1] or ema20.iloc[-1] <= ema20.iloc[-3]: return None if cur < ema20.iloc[-1] or cur > ema20.iloc[-1] * 1.04: return None if cur_vol_ratio < 0.85: return None if cur_hist <= 0 and macd_line.iloc[-1] <= signal_line.iloc[-1]: return None score = 0 reasons = [] if ema5.iloc[-1] > ema20.iloc[-1] > ema60.iloc[-1]: score += 3 reasons.append("메이저정배열") elif cur > ema20.iloc[-1] > ema60.iloc[-1]: score += 2 reasons.append("중기상승") if ema5.iloc[-1] > ema5.iloc[-3] and cur >= ema5.iloc[-1]: score += 2 reasons.append("단기회복") elif close.iloc[-1] > close.iloc[-2]: score += 1 reasons.append("양봉회복") if macd_line.iloc[-1] > signal_line.iloc[-1] and cur_hist > prev_hist and cur_hist > 0: score += 3 reasons.append("MACD확장") elif cur_hist > 0 and cur_hist > prev_hist: score += 2 reasons.append("MACD상승") elif cur_hist > 0: score += 1 reasons.append("MACD양수") else: return None if 48 <= cur_rsi <= 60: score += 2 reasons.append(f"RSI건강({cur_rsi:.0f})") else: score += 1 reasons.append(f"RSI안정({cur_rsi:.0f})") if cur_vol_ratio >= 1.6: score += 2 reasons.append(f"거래량확대({cur_vol_ratio:.1f}x)") elif cur_vol_ratio >= 1.1: score += 1 reasons.append(f"거래량확인({cur_vol_ratio:.1f}x)") recent_high = close.iloc[-20:].max() if cur >= recent_high * 0.985: score += 1 reasons.append("고점근접") stop_loss = min(max(atr_pct * 1.15, 0.9), 1.9) target_profit = min(max(stop_loss * 1.9, 1.6), 3.4) return _build_signal_result( "major", cur, score, reasons, target_profit, stop_loss, atr_pct, float(cur_rsi), float(cur_vol_ratio), ) def analyze_alt_coin(df: pd.DataFrame) -> dict | None: if df is None or len(df) < MIN_SIGNAL_BARS: return None close = df["close"] volume = df["volume"] cur = close.iloc[-1] profile = STRATEGY_PROFILES["alt"] rsi = calc_rsi(close) macd_line, signal_line, macd_hist = calc_macd(close) ema5 = calc_ema(close, 5) ema20 = calc_ema(close, 20) ema60 = calc_ema(close, 60) atr = calc_atr(df) vol_avg = volume.rolling(20).mean() cur_rsi = rsi.iloc[-1] cur_hist = macd_hist.iloc[-1] prev_hist = macd_hist.iloc[-2] cur_atr = atr.iloc[-1] cur_vol_avg = vol_avg.iloc[-1] cur_vol_ratio = (volume.iloc[-1] / cur_vol_avg) if cur_vol_avg > 0 else 0 if ( pd.isna(cur_rsi) or pd.isna(cur_hist) or pd.isna(prev_hist) or pd.isna(cur_atr) or cur_atr <= 0 or pd.isna(cur_vol_avg) ): return None atr_pct = (cur_atr / cur) * 100 if atr_pct < profile["min_atr_pct"] or atr_pct > profile["max_atr_pct"]: return None if cur_rsi < 48 or cur_rsi > 74: return None if not (ema5.iloc[-1] > ema20.iloc[-1] > ema60.iloc[-1]): return None if ema20.iloc[-1] <= ema20.iloc[-3]: return None if cur < ema20.iloc[-1] or cur > ema20.iloc[-1] * 1.08: return None if cur_hist <= 0 or cur_hist < prev_hist: return None if cur_vol_ratio < 1.35: return None recent_breakout = close.iloc[-21:-1].max() if cur < recent_breakout * 0.992: return None score = 5 reasons = ["돌파기반"] if cur >= recent_breakout: score += 2 reasons.append("직전고점돌파") else: score += 1 reasons.append("고점압박") if macd_line.iloc[-2] <= signal_line.iloc[-2] and macd_line.iloc[-1] > signal_line.iloc[-1]: score += 2 reasons.append("MACD돌파") elif macd_line.iloc[-1] > signal_line.iloc[-1] and cur_hist > prev_hist: score += 2 reasons.append("MACD확장") else: score += 1 reasons.append("MACD양수") if cur_vol_ratio >= 1.9: score += 2 reasons.append(f"거래량폭증({cur_vol_ratio:.1f}x)") else: score += 1 reasons.append(f"거래량동반({cur_vol_ratio:.1f}x)") if 55 <= cur_rsi <= 68: score += 2 reasons.append(f"RSI가속({cur_rsi:.0f})") else: score += 1 reasons.append(f"RSI유지({cur_rsi:.0f})") candle_range = max(df["high"].iloc[-1] - df["low"].iloc[-1], cur * 0.0001) close_strength = (cur - df["low"].iloc[-1]) / candle_range if close_strength >= 0.7 and cur > df["open"].iloc[-1]: score += 1 reasons.append("종가강세") stop_loss = min(max(atr_pct * 1.4, 1.4), 3.0) target_profit = min(max(stop_loss * 2.0, 2.8), 5.2) return _build_signal_result( "alt", cur, score, reasons, target_profit, stop_loss, atr_pct, float(cur_rsi), float(cur_vol_ratio), ) def analyze_coin(df: pd.DataFrame, ticker: str | None = None) -> dict | None: strategy_kind = get_strategy_kind(ticker) if strategy_kind == "major": return analyze_major_coin(df) return analyze_alt_coin(df) def check_sell_signal( df: pd.DataFrame, buy_price: float, target: float, stop: float, bought_at: str = None, held_candles: int | None = None, strategy_kind: str = "major", ) -> tuple[bool, str]: if df is None or len(df) < 60: return False, "" close = df["close"] cur = close.iloc[-1] profit_rate = ((cur - buy_price) / buy_price) * 100 ema5 = calc_ema(close, 5) ema20 = calc_ema(close, 20) _, _, macd_hist = calc_macd(close) atr = calc_atr(df) atr_pct = (atr.iloc[-1] / cur) * 100 if cur > 0 else 0 rsi = calc_rsi(close) if held_candles is None and bought_at: bought_at_dt = parse_trade_time(bought_at) if bought_at_dt: held_minutes = (datetime.now() - bought_at_dt).total_seconds() / 60 held_candles = max(int(held_minutes // 15), 0) if profit_rate <= -stop: return True, f"손절({profit_rate:+.2f}%)" if profit_rate >= target: return True, f"목표수익({profit_rate:+.2f}%)" if strategy_kind == "major": trail_trigger = max(target * 0.45, 0.8) trail_floor = max(atr_pct * 0.9, 0.8) trail_window = 14 time_cut_1 = 24 time_cut_2 = 48 else: trail_trigger = max(target * 0.35, 1.0) trail_floor = max(atr_pct * 0.85, 1.0) trail_window = 8 time_cut_1 = 12 time_cut_2 = 24 if profit_rate >= trail_trigger: if held_candles is not None: trail_window = min(max(held_candles + 1, 6), trail_window) recent_high = close.iloc[-trail_window:].max() drop_from_high = ((recent_high - cur) / recent_high) * 100 if drop_from_high >= trail_floor: return True, f"트레일링스탑({profit_rate:+.2f}%)" trend_break_floor = 0.6 if strategy_kind == "major" else 0.8 if profit_rate > trend_break_floor and cur < ema5.iloc[-1] and macd_hist.iloc[-1] < macd_hist.iloc[-2]: return True, f"단기추세이탈({profit_rate:+.2f}%)" overheat_rsi = 76 if strategy_kind == "major" else 74 overheat_floor = 1.0 if strategy_kind == "major" else 1.5 if rsi.iloc[-1] > overheat_rsi and macd_hist.iloc[-1] < macd_hist.iloc[-2] and profit_rate > overheat_floor: return True, f"과열둔화({profit_rate:+.2f}%)" if held_candles is not None: time_floor = 0.3 if strategy_kind == "major" else 0.6 time_ema = ema20.iloc[-1] if strategy_kind == "major" else ema5.iloc[-1] if held_candles >= time_cut_1 and profit_rate < time_floor and cur < time_ema: return True, f"시간정리({held_candles * 15}분, {profit_rate:+.2f}%)" if held_candles >= time_cut_2 and profit_rate < 0: return True, f"장기부진({held_candles * 15}분, {profit_rate:+.2f}%)" return False, "" def get_cooldown_bars(ticker: str, profit_krw: float) -> int: profile = get_strategy_profile(ticker) if profit_krw < 0: return profile["cooldown_loss"] return profile["cooldown_win"] # ============================================================ # Trader # ============================================================ class UpbitTrader: MIN_BUY = 5500 # 업비트 최소 주문금액 + 여유 FEE_RATE = 0.0005 # 업비트 수수료 0.05% (매수+매도 = 0.1%) def __init__(self): access_key = os.environ.get("UPBIT_ACCESS_KEY", "") secret_key = os.environ.get("UPBIT_SECRET_KEY", "") if access_key and secret_key: self.upbit = pyupbit.Upbit(access_key, secret_key) else: self.upbit = None self.running = False self.total_budget = 0 # 총 투자금 self.total_realized = 0.0 # 누적 실현 수익 self.total_trades = 0 self.started_at: str | None = None # {ticker: position} self.positions: dict[str, dict] = {} self._last_scan: float = 0 self._last_report: float = 0 self._candidates: list[dict] = [] self._report_interval = 600 # 10분 포지션 리포트 self.compound = False # 복리 모드 self._http_session: aiohttp.ClientSession | None = None self._allocations: dict[str, float] = {} # 종목별 투자금 배분 self._strategy = "" # 현재 전략 라벨 self._market_trend = "neutral" self._last_scan_candle: datetime | None = None self._last_scan_at: datetime | None = None self._last_scan_stats: dict[str, int] = {} self._last_scan_analyzed: list[str] = [] self._daily_screen_cache: dict[str, dict | None] = {} self._daily_screen_cache_time: float = 0 self._daily_screen_cache_updated_at: datetime | None = None self._last_sell_candle: datetime | None = None self._cooldowns: dict[str, datetime] = {} # 가상머니 모드 self.paper_mode = False self._virtual_krw = 0.0 self._virtual_coins: dict[str, float] = {} # {coin: volume} def is_configured(self) -> bool: if self.paper_mode: return True return self.upbit is not None @property def effective_budget(self) -> float: """복리 모드: 수익금 포함 / 일반: 초기 투자금""" if self.compound: return max(self.total_budget + self.total_realized, self.MIN_BUY) return self.total_budget @property def max_coins(self) -> int: """예산 기반 최대 보유 종목 수 (동적)""" budget = self.effective_budget if budget <= 0: return 1 # 코인당 최소 1만원, 최대 5종목 return min(int(budget // 10000), 5) @property def invest_per_coin(self) -> float: """균등 배분 (폴백용)""" return self.effective_budget / self.max_coins if self.max_coins > 0 else 0 @property def remaining_budget(self) -> float: """남은 투자 가능 예산 = 총 예산 - 이미 투자중인 금액""" invested = sum( (p["buy_price"] * p["buy_volume"]) for p in self.positions.values() if p["status"] == "holding" and p.get("buy_price") and p.get("buy_volume") ) return max(self.effective_budget - invested, 0) def _calc_allocations(self, candidates: list[dict], avail_budget: float) -> dict[str, float]: """전략적 종목 수 결정 + 점수 비례 투자금 가중 배분 - 고확신 시그널(평균 7+): 1-2종목 집중 - 중간 시그널(5-6): 2-3종목 분산 - 약한 시그널(4): 최대한 분산 """ if not candidates: return {} if avail_budget < self.MIN_BUY: self._strategy = "잔고 부족" return {} holding = len([p for p in self.positions.values() if p["status"] in ("holding", "buying")]) max_new = self.max_coins - holding if max_new <= 0: self._strategy = "슬롯 없음" return {} available = [c for c in candidates if c["ticker"] not in self.positions] if not available: self._strategy = "후보 없음" return {} # 전략적 종목 수 결정 top_n = available[:max_new] avg_score = sum(c["score"] for c in top_n) / len(top_n) if avg_score >= 8: target = min(2, max_new, len(available)) self._strategy = f"집중투자({avg_score:.1f}점)" elif avg_score >= 6: target = min(3, max_new, len(available)) self._strategy = f"균형분산({avg_score:.1f}점)" else: target = min(max_new, len(available)) self._strategy = f"광역분산({avg_score:.1f}점)" # 최소 금액 보장 조정 while target > 1 and avail_budget / target < self.MIN_BUY: target -= 1 selected = available[:target] if not selected: return {} # 점수 비례 가중 배분 total_score = sum(c["score"] for c in selected) if total_score <= 0: per = avail_budget / len(selected) return {c["ticker"]: per for c in selected} allocations = {} remaining = avail_budget for i, c in enumerate(selected): if i == len(selected) - 1: amount = remaining else: amount = round(avail_budget * c["score"] / total_score, -2) amount = max(amount, self.MIN_BUY) if amount >= self.MIN_BUY and remaining >= self.MIN_BUY: allocations[c["ticker"]] = amount remaining -= amount return allocations # --- API wrappers --- async def get_balance(self, currency: str = "KRW") -> float: if self.paper_mode: if currency == "KRW": return self._virtual_krw return self._virtual_coins.get(currency, 0.0) if not self.upbit: return 0 return await asyncio.to_thread(self.upbit.get_balance, currency) async def get_balances(self) -> list: if self.paper_mode: balances = [{"currency": "KRW", "balance": str(self._virtual_krw), "locked": "0", "avg_buy_price": "0"}] for coin, vol in self._virtual_coins.items(): if vol > 0: pos = self.positions.get(f"KRW-{coin}") avg = pos["buy_price"] if pos and pos.get("buy_price") else 0 balances.append({"currency": coin, "balance": str(vol), "locked": "0", "avg_buy_price": str(avg)}) return balances if not self.upbit: return [] return await asyncio.to_thread(self.upbit.get_balances) # --- API helpers --- async def _ensure_session(self) -> aiohttp.ClientSession: if self._http_session is None or self._http_session.closed: self._http_session = aiohttp.ClientSession() return self._http_session async def _get_volume_ranking(self, tickers: list[str]) -> dict[str, float]: """24시간 거래대금 조회""" if not tickers: return {} session = await self._ensure_session() volume_map = {} chunk_size = 50 for i in range(0, len(tickers), chunk_size): chunk = tickers[i:i + chunk_size] try: async with session.get( "https://api.upbit.com/v1/ticker", params={"markets": ",".join(chunk)} ) as resp: if resp.status == 200: data = await resp.json() for item in data: volume_map[item["market"]] = item.get("acc_trade_price_24h", 0) except Exception: continue if i + chunk_size < len(tickers): await asyncio.sleep(0.1) return volume_map async def _fetch_current_prices(self, tickers: list[str]) -> dict[str, float]: """현재가 일괄 조회""" if not tickers: return {} session = await self._ensure_session() prices = {} try: async with session.get( "https://api.upbit.com/v1/ticker", params={"markets": ",".join(tickers)} ) as resp: if resp.status == 200: data = await resp.json() for item in data: prices[item["market"]] = item["trade_price"] except Exception: pass return prices async def _get_order_flow(self, ticker: str) -> dict | None: """공개 Upbit API로 최근 체결과 상위 호가 수급을 조회한다.""" session = await self._ensure_session() try: async with session.get( "https://api.upbit.com/v1/orderbook", params={"markets": ticker, "count": 15}, ) as response: if response.status != 200: return None orderbooks = await response.json() if not orderbooks: return None await asyncio.sleep(0.12) async with session.get( "https://api.upbit.com/v1/trades/ticks", params={"market": ticker, "count": ORDER_FLOW_TRADE_COUNT}, ) as response: if response.status != 200: return None trades = await response.json() await asyncio.sleep(0.12) except Exception as exc: logger.debug("Order flow lookup failed [%s]: %s", ticker, exc) return None return analyze_order_flow(orderbooks[0], trades) # --- Scan --- def request_immediate_rescan(self) -> None: """다음 자동매매 주기에서 캔들 게이트를 건너뛰고 다시 스캔한다.""" self._last_scan = 0 self._last_scan_candle = None async def _refresh_daily_screen_cache( self, tickers: list[str], ) -> int: """전체 KRW 종목의 일봉 추세와 ATR을 제한 속도 내에서 갱신한다.""" async def fetch_one(ticker: str): try: df_daily = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="day", count=DAILY_FETCH_COUNT, ) df_daily = get_completed_candle_df( df_daily, interval_minutes=1440 ) if df_daily is None or len(df_daily) < DAILY_TREND_BARS: return ticker, None, False daily_window = df_daily.tail(DAILY_TREND_BARS) trend, _ = check_trend(daily_window) daily_atr_pct = calc_atr_pct(daily_window) return ticker, { "trend": trend, "daily_atr_pct": daily_atr_pct, }, False except Exception as exc: logger.debug("Daily screen lookup failed [%s]: %s", ticker, exc) return ticker, None, True cache: dict[str, dict | None] = {} errors = 0 for start in range(0, len(tickers), DAILY_SCREEN_BATCH_SIZE): batch = tickers[start:start + DAILY_SCREEN_BATCH_SIZE] results = await asyncio.gather(*(fetch_one(ticker) for ticker in batch)) for ticker, result, failed in results: cache[ticker] = result if failed: errors += 1 if start + DAILY_SCREEN_BATCH_SIZE < len(tickers): await asyncio.sleep(DAILY_SCREEN_BATCH_DELAY_SECONDS) self._daily_screen_cache = cache self._daily_screen_cache_time = asyncio.get_event_loop().time() self._daily_screen_cache_updated_at = datetime.now() logger.info( "Daily market cache refreshed: markets=%d errors=%d", len(tickers), errors, ) return errors async def _ensure_daily_screen_cache( self, tickers: list[str], ) -> tuple[bool, int]: now = asyncio.get_event_loop().time() cache_age = now - self._daily_screen_cache_time missing_tickers = any( ticker not in self._daily_screen_cache for ticker in tickers ) should_refresh = ( not self._daily_screen_cache or cache_age >= DAILY_SCREEN_CACHE_SECONDS or missing_tickers ) if not should_refresh: return False, 0 errors = await self._refresh_daily_screen_cache(tickers) return True, errors async def _check_btc_trend(self) -> tuple[str, int]: """BTC 일봉 추세. 하락 시 점수 감점, 상승 시 보너스.""" try: df_btc = await asyncio.to_thread( pyupbit.get_ohlcv, "KRW-BTC", interval="day", count=DAILY_FETCH_COUNT ) df_btc = get_completed_candle_df(df_btc, interval_minutes=1440) if df_btc is None or len(df_btc) < 50: return "neutral", 0 trend, bonus = check_trend(df_btc.tail(DAILY_TREND_BARS)) return trend, bonus except Exception: return "neutral", 0 async def scan_market(self) -> list[dict]: tickers = await asyncio.to_thread(pyupbit.get_tickers, fiat="KRW") if not tickers: return [] cache_refreshed, cache_errors = await self._ensure_daily_screen_cache(tickers) btc_daily = self._daily_screen_cache.get("KRW-BTC") if btc_daily: btc_trend = btc_daily["trend"] else: btc_trend, _ = await self._check_btc_trend() self._market_trend = btc_trend available = [t for t in tickers if t not in self.positions] volume_map = await self._get_volume_ranking(available) market_tickers = rank_market_tickers(available, volume_map) candidates = [] analyzed_coins = [] scan_stats = Counter( universe=len(tickers), scanned=len(market_tickers), daily_cache_refreshed=int(cache_refreshed), daily_cache_errors=cache_errors, ) for ticker in market_tickers: try: cooldown_until = self._cooldowns.get(ticker) if cooldown_until and cooldown_until > datetime.now(): scan_stats["cooldown"] += 1 continue profile = get_strategy_profile(ticker) daily_screen = self._daily_screen_cache.get(ticker) if daily_screen is None: scan_stats["daily_data"] += 1 continue trend = daily_screen["trend"] trend_adjustment = get_entry_trend_adjustment( ticker, trend, btc_trend ) if trend_adjustment is None: scan_stats[f"trend_{trend}"] += 1 continue daily_atr_pct = daily_screen["daily_atr_pct"] if ( daily_atr_pct is None or daily_atr_pct < profile["min_daily_atr_pct"] or daily_atr_pct > profile["max_daily_atr_pct"] ): scan_stats["daily_atr"] += 1 continue df = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="minute15", count=100 ) df = get_completed_candle_df(df, interval_minutes=15) await asyncio.sleep(0.12) if df is None or len(df) < MIN_SIGNAL_BARS: scan_stats["signal_data"] += 1 continue analyzed_coins.append(ticker.replace("KRW-", "")) result = analyze_coin(df, ticker) if result: order_flow = await self._get_order_flow(ticker) if order_flow is None: scan_stats["order_flow_data"] += 1 continue if order_flow["blocked"]: scan_stats["order_flow_blocked"] += 1 continue result["score"] += trend_adjustment result["score"] += order_flow["score"] result["reasons"].extend(order_flow["reasons"]) result["trend"] = trend result["btc_trend"] = btc_trend result["daily_atr_pct"] = round(daily_atr_pct, 2) result["order_flow"] = order_flow if result and result["score"] >= profile["entry_threshold"]: scan_stats["candidates"] += 1 candidates.append({ "ticker": ticker, "coin": ticker.replace("KRW-", ""), "vol_24h": volume_map.get(ticker, 0), "df": df, **result, }) elif result: scan_stats["entry_score"] += 1 else: scan_stats["signal_filter"] += 1 except Exception as exc: scan_stats["errors"] += 1 logger.debug("Scan skipped [%s]: %s", ticker, exc) continue # 점수 우선, 동점시 거래대금 우선 candidates.sort(key=lambda x: (x["score"], x["vol_24h"]), reverse=True) scan_stats["candidates"] = len(candidates) self._last_scan_at = datetime.now() self._last_scan_stats = dict(sorted(scan_stats.items())) self._last_scan_analyzed = analyzed_coins if not candidates: self._strategy = ( f"KRW 전체 {len(market_tickers)}종목 스캔 | " f"시장기준 BTC {btc_trend} | 상대강도 신호 대기" ) logger.info( "Market scan completed: market_btc=%s analyzed=%s stats=%s", btc_trend, analyzed_coins, self._last_scan_stats, ) return candidates # --- Trade Cycle --- async def execute_cycle(self, send_fn) -> None: if not self.running: return now = asyncio.get_event_loop().time() signal_candle = get_latest_closed_candle_start(15) if self._last_scan_candle != signal_candle: self._last_scan = now self._last_scan_candle = signal_candle try: self._candidates = await self.scan_market() if self.paper_mode: krw = self._virtual_krw else: krw = await self.get_balance("KRW") budget_limit = min(krw, self.remaining_budget) self._allocations = self._calc_allocations(self._candidates, budget_limit) holding = len([p for p in self.positions.values() if p["status"] in ("holding", "buying")]) slots = self.max_coins - holding if self._candidates: top = self._candidates[:5] btc_label = ( f" | 시장BTC:{self._market_trend}" if self._market_trend != "neutral" else "" ) lines = [f"[마켓 스캔] {len(self._candidates)}개 후보 | 전략: {self._strategy}{btc_label}\n"] for i, c in enumerate(top, 1): alloc = self._allocations.get(c["ticker"]) alloc_str = f" -> {alloc:,.0f}원" if alloc else "" lines.append( f"{i}. {c['coin']} | {c['price']:,.0f}원{alloc_str}\n" f" 점수: {c['score']}점 | RSI: {c['rsi']:.0f} | 거래대금: {c.get('vol_24h', 0) / 100000000:.0f}억\n" f" 전략: {c.get('strategy_name', '?')} | 목표 +{c['target_profit']}% / 손절 -{c['stop_loss']}% | 일봉ATR {c.get('daily_atr_pct', 0):.1f}%\n" f" 수급: 호가매수 {c['order_flow']['book_buy_ratio']:.0%} | " f"체결매수 {c['order_flow']['trade_buy_ratio']:.0%} | " f"대량매수 {c['order_flow']['large_buy_ratio']:.0%} | " f"스프레드 {c['order_flow']['spread_bps']:.1f}bp\n" f" 추세: {c.get('trend', '?')} | 신호: {', '.join(c['reasons'])}" ) if self._allocations: buy_list = [f"{c['coin']}({self._allocations[c['ticker']]:,.0f}원)" for c in self._candidates if c["ticker"] in self._allocations] lines.append(f"\n→ 매수 예정 ({len(buy_list)}종목): {', '.join(buy_list)}") elif slots > 0: lines.append(f"\n→ 잔고 부족 (가용: {krw:,.0f}원)") else: lines.append(f"\n→ 슬롯 없음, 매도 대기 중") await send_fn("\n".join(lines)) except Exception as e: logger.error(f"Scan error: {e}") # 정기 포지션 리포트 (10분마다) if self.positions and now - self._last_report >= self._report_interval: self._last_report = now try: lines = [f"[포지션 리포트] {datetime.now().strftime('%H:%M')}\n"] total_unrealized = 0 for pos in self.positions.values(): if pos["status"] == "holding" and pos["buy_price"]: cur = await asyncio.to_thread(pyupbit.get_current_price, pos["ticker"]) if cur: rate = ((cur - pos["buy_price"]) / pos["buy_price"]) * 100 pnl = (cur - pos["buy_price"]) * pos["buy_volume"] total_unrealized += pnl lines.append( f" {pos['coin']}: {rate:+.2f}% ({pnl:+,.0f}원)\n" f" {pos['buy_price']:,.0f} → {cur:,.0f}원\n" f" 목표: +{pos['target_profit']}% | 손절: -{pos['stop_loss']}%" ) lines.append(f"\n미실현: {total_unrealized:+,.0f}원") lines.append(f"실현: {self.total_realized:+,.0f}원") lines.append(f"총손익: {self.total_realized + total_unrealized:+,.0f}원") await send_fn("\n".join(lines)) except Exception as e: logger.error(f"Report error: {e}") do_sell_check = self._last_sell_candle != signal_candle if do_sell_check: self._last_sell_candle = signal_candle for ticker, pos in list(self.positions.items()): if pos["status"] != "holding": continue if not do_sell_check: continue try: df = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="minute15", count=100 ) df = get_completed_candle_df(df, interval_minutes=15) if df is None or len(df) < MIN_SIGNAL_BARS: continue should_sell, reason = check_sell_signal( df, pos["buy_price"], pos["target_profit"], pos["stop_loss"], bought_at=pos.get("bought_at"), held_candles=max( int((datetime.now() - parse_trade_time(pos.get("bought_at"))).total_seconds() // 900), 0, ) if parse_trade_time(pos.get("bought_at")) else None, strategy_kind=get_strategy_kind(ticker), ) if should_sell: await self._sell(pos, reason, send_fn) except Exception as e: logger.error(f"Sell check [{ticker}]: {e}") await asyncio.sleep(0.12) # 배분된 종목 매수 if self._allocations: for ticker, amount in list(self._allocations.items()): if ticker not in self.positions: cand = next((c for c in self._candidates if c["ticker"] == ticker), None) if cand: await self._buy(cand, send_fn, amount) self._allocations.pop(ticker, None) await asyncio.sleep(0.5) async def _buy(self, cand: dict, send_fn, amount: float | None = None) -> None: ticker = cand["ticker"] coin = cand["coin"] if amount is None: amount = self.invest_per_coin # 예산 상한 안전장치: invest_per_coin의 1.5배 초과 방지 max_per_coin = self.invest_per_coin * 1.5 amount = min(amount, max_per_coin, self.remaining_budget) if amount < self.MIN_BUY: return krw = await self.get_balance("KRW") if krw < amount: return self.positions[ticker] = { "coin": coin, "ticker": ticker, "status": "buying", "buy_price": None, "buy_volume": None, "bought_at": None, "target_profit": cand["target_profit"], "stop_loss": cand["stop_loss"], "strategy_kind": cand.get("strategy_kind"), "strategy_name": cand.get("strategy_name"), } if self.paper_mode: # 가상 매수: 현재가로 즉시 체결 buy_price = cand["price"] fee = amount * self.FEE_RATE actual_amount = amount - fee volume = actual_amount / buy_price self._virtual_krw -= amount self._virtual_coins[coin] = self._virtual_coins.get(coin, 0) + volume self.positions[ticker].update({ "status": "holding", "buy_price": buy_price, "buy_volume": volume, "bought_at": format_trade_time(), }) self.total_trades += 1 await save_session(self) await save_position(self.positions[ticker]) target_price = buy_price * (1 + cand['target_profit'] / 100) stop_price = buy_price * (1 - cand['stop_loss'] / 100) chart = render_chart( cand.get("df"), title=f"{coin} Buy (PAPER)", buy_price=buy_price, ) await send_fn( f"[가상 매수] {coin} (#{self.total_trades})\n" f"━━━━━━━━━━━━━━━\n" f"매수가: {buy_price:,.0f}원\n" f"투자금: {amount:,.0f}원\n" f"━━━━━━━━━━━━━━━\n" f"목표가: {target_price:,.0f}원 (+{cand['target_profit']}%)\n" f"손절가: {stop_price:,.0f}원 (-{cand['stop_loss']}%)\n" f"━━━━━━━━━━━━━━━\n" f"전략: {cand.get('strategy_name', '?')}\n" f"매수 근거: {', '.join(cand['reasons'])}\n" f"포지션: {len(self.positions)}/{self.max_coins}", photo=chart, ) return result = await asyncio.to_thread(self.upbit.buy_market_order, ticker, amount) if result and isinstance(result, dict) and "error" not in result: await asyncio.sleep(0.5) volume = await self.get_balance(coin) avg_price = await asyncio.to_thread(self.upbit.get_avg_buy_price, coin) self.positions[ticker].update({ "status": "holding", "buy_price": avg_price or cand["price"], "buy_volume": volume, "bought_at": format_trade_time(), }) self.total_trades += 1 # DB 저장 await save_session(self) await save_position(self.positions[ticker]) buy_price = avg_price or cand['price'] target_price = buy_price * (1 + cand['target_profit'] / 100) stop_price = buy_price * (1 - cand['stop_loss'] / 100) # 캔들 차트 생성 chart = render_chart( cand.get("df"), title=f"{coin} Buy", buy_price=buy_price, ) await send_fn( f"[매수 체결] {coin} (#{self.total_trades})\n" f"━━━━━━━━━━━━━━━\n" f"매수가: {buy_price:,.0f}원\n" f"투자금: {amount:,.0f}원\n" f"━━━━━━━━━━━━━━━\n" f"목표가: {target_price:,.0f}원 (+{cand['target_profit']}%)\n" f"손절가: {stop_price:,.0f}원 (-{cand['stop_loss']}%)\n" f"━━━━━━━━━━━━━━━\n" f"전략: {cand.get('strategy_name', '?')}\n" f"매수 근거: {', '.join(cand['reasons'])}\n" f"포지션: {len(self.positions)}/{self.max_coins}", photo=chart, ) else: del self.positions[ticker] err = result.get("error", {}).get("message", str(result)) if isinstance(result, dict) else str(result) logger.error(f"Buy failed [{coin}]: {err}") async def _sell(self, pos: dict, reason: str, send_fn) -> None: ticker = pos["ticker"] coin = pos["coin"] pos["status"] = "selling" if self.paper_mode: # 가상 매도 sell_volume = self._virtual_coins.get(coin, 0) if sell_volume <= 0: del self.positions[ticker] return cur_price = await asyncio.to_thread(pyupbit.get_current_price, ticker) if not cur_price: pos["status"] = "holding" return profit_rate = ((cur_price - pos["buy_price"]) / pos["buy_price"]) * 100 gross = (cur_price - pos["buy_price"]) * sell_volume buy_fee = pos["buy_price"] * sell_volume * self.FEE_RATE sell_fee = cur_price * sell_volume * self.FEE_RATE profit_krw = gross - buy_fee - sell_fee self._virtual_krw += cur_price * sell_volume - sell_fee self._virtual_coins.pop(coin, None) self.total_realized += profit_krw cooldown_bars = get_cooldown_bars(ticker, profit_krw) self._cooldowns[ticker] = datetime.now() + timedelta(minutes=15 * cooldown_bars) del self.positions[ticker] await save_session(self) await delete_position(ticker) remaining_cands = [ c for c in self._candidates if c["ticker"] != ticker and c["ticker"] not in self.positions ] if remaining_cands: budget_limit = min(self._virtual_krw, self.remaining_budget) self._allocations = self._calc_allocations(remaining_cands, budget_limit) next_action = f"-> 즉시 재투자 {len(self._allocations)}종목 배분" else: self.request_immediate_rescan() next_action = "-> 다음 자동매매 주기에 즉시 리스캔" try: chart_df = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="minute15", count=100 ) chart = render_chart( chart_df, title=f"{coin} Sell (PAPER)", buy_price=pos.get("buy_price"), sell_price=cur_price, ) except Exception: chart = None result_emoji = "익절" if profit_krw >= 0 else "손절" await send_fn( f"[가상 매도] {coin} - {result_emoji}\n" f"━━━━━━━━━━━━━━━\n" f"매도 사유: {reason}\n" f"매수가: {pos['buy_price']:,.0f}원 -> 매도가: {cur_price:,.0f}원\n" f"수익률: {profit_rate:+.2f}%\n" f"수익금: {profit_krw:+,.0f}원 (수수료 차감)\n" f"━━━━━━━━━━━━━━━\n" f"누적수익: {self.total_realized:+,.0f}원\n" f"가상잔고: {self._virtual_krw:,.0f}원\n" f"총거래: {self.total_trades}회\n" f"{next_action}", photo=chart, ) return # 실제 잔고 기반 매도 (기억된 수량과 다를 수 있음) actual_balance = await self.get_balance(coin) if actual_balance <= 0: del self.positions[ticker] await send_fn(f"[매도 스킵] {coin} - 잔고 없음 (이미 체결됨)") return sell_volume = actual_balance cur_price = await asyncio.to_thread(pyupbit.get_current_price, ticker) result = await asyncio.to_thread(self.upbit.sell_market_order, ticker, sell_volume) if result and isinstance(result, dict) and "error" not in result: # 체결 확인 대기 await asyncio.sleep(0.5) remaining = await self.get_balance(coin) if remaining > 0 and remaining > sell_volume * 0.01: # 미체결 잔량 있으면 재시도 retry = await asyncio.to_thread(self.upbit.sell_market_order, ticker, remaining) if retry and isinstance(retry, dict) and "error" not in retry: await asyncio.sleep(0.3) remaining = await self.get_balance(coin) profit_krw = 0.0 profit_rate = 0.0 if cur_price and pos["buy_price"]: profit_rate = ((cur_price - pos["buy_price"]) / pos["buy_price"]) * 100 gross = (cur_price - pos["buy_price"]) * sell_volume buy_fee = pos["buy_price"] * sell_volume * self.FEE_RATE sell_fee = cur_price * sell_volume * self.FEE_RATE profit_krw = gross - buy_fee - sell_fee self.total_realized += profit_krw cooldown_bars = get_cooldown_bars(ticker, profit_krw) self._cooldowns[ticker] = datetime.now() + timedelta(minutes=15 * cooldown_bars) del self.positions[ticker] # DB 저장 await save_session(self) await delete_position(ticker) # 매도 후 즉시 재배분: 남은 후보로 재계산 or 리스캔 remaining_cands = [ c for c in self._candidates if c["ticker"] != ticker and c["ticker"] not in self.positions ] if remaining_cands: new_krw = await self.get_balance("KRW") budget_limit = min(new_krw, self.remaining_budget) self._allocations = self._calc_allocations(remaining_cands, budget_limit) next_action = f"-> 즉시 재투자 {len(self._allocations)}종목 배분" else: self.request_immediate_rescan() next_action = "-> 다음 자동매매 주기에 즉시 리스캔" # 매도 차트 생성 try: chart_df = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="minute15", count=100 ) chart = render_chart( chart_df, title=f"{coin} Sell", buy_price=pos.get("buy_price"), sell_price=cur_price, ) except Exception: chart = None status = " (일부 미체결)" if remaining > sell_volume * 0.01 else "" result_emoji = "익절" if profit_krw >= 0 else "손절" await send_fn( f"[매도 체결] {coin} - {result_emoji}{status}\n" f"━━━━━━━━━━━━━━━\n" f"매도 사유: {reason}\n" f"매수가: {pos['buy_price']:,.0f}원 -> 매도가: {cur_price:,.0f}원\n" f"수익률: {profit_rate:+.2f}%\n" f"수익금: {profit_krw:+,.0f}원 (수수료 차감)\n" f"━━━━━━━━━━━━━━━\n" f"누적수익: {self.total_realized:+,.0f}원\n" f"총거래: {self.total_trades}회\n" f"{next_action}", photo=chart, ) else: pos["status"] = "holding" pos["_sell_retry"] = pos.get("_sell_retry", 0) + 1 err = result.get("error", {}).get("message", str(result)) if isinstance(result, dict) else str(result) logger.error(f"Sell failed [{coin}] (retry {pos['_sell_retry']}): {err}") # 3회 실패 시 알림 if pos["_sell_retry"] >= 3: await send_fn(f"[매도 실패] {coin} - {err}\n수동 확인 필요") pos["_sell_retry"] = 0 # --- Manual Controls --- async def force_sell(self, coin: str) -> tuple[bool, str]: ticker = f"KRW-{coin.upper()}" pos = self.positions.get(ticker) if not pos: return False, f"{coin} 보유 포지션 없음" if pos["status"] != "holding": return False, f"{coin} {pos['status']} 상태" if self.paper_mode: vol = self._virtual_coins.get(coin.upper(), 0) cur = await asyncio.to_thread(pyupbit.get_current_price, ticker) profit_krw = 0.0 if cur and pos["buy_price"] and vol > 0: gross = (cur - pos["buy_price"]) * vol buy_fee = pos["buy_price"] * vol * self.FEE_RATE sell_fee = cur * vol * self.FEE_RATE profit_krw = gross - buy_fee - sell_fee self._virtual_krw += cur * vol - sell_fee self._virtual_coins.pop(coin.upper(), None) self.total_realized += profit_krw del self.positions[ticker] label = "[가상 수동매도]" return True, f"{label} {coin}\n수익금: {profit_krw:+,.0f}원 (수수료 차감)" actual_balance = await self.get_balance(coin.upper()) if actual_balance <= 0: del self.positions[ticker] return True, f"{coin} 잔고 없음 (이미 매도됨)" cur = await asyncio.to_thread(pyupbit.get_current_price, ticker) result = await asyncio.to_thread(self.upbit.sell_market_order, ticker, actual_balance) if result and isinstance(result, dict) and "error" not in result: profit_krw = 0.0 if cur and pos["buy_price"]: gross = (cur - pos["buy_price"]) * actual_balance buy_fee = pos["buy_price"] * actual_balance * self.FEE_RATE sell_fee = cur * actual_balance * self.FEE_RATE profit_krw = gross - buy_fee - sell_fee self.total_realized += profit_krw del self.positions[ticker] return True, f"[수동매도] {coin}\n수익금: {profit_krw:+,.0f}원 (수수료 차감)" return False, f"매도 실패: {result}" async def force_sell_all(self) -> list[str]: results = [] for pos in list(self.positions.values()): if pos["status"] == "holding": _, msg = await self.force_sell(pos["coin"]) results.append(msg) await asyncio.sleep(0.3) return results async def get_monitor_text(self) -> str | None: """1초 주기 실시간 모니터링 텍스트""" holding = [p for p in self.positions.values() if p["status"] == "holding" and p["buy_price"]] if not holding: return None tickers = [p["ticker"] for p in holding] prices = await self._fetch_current_prices(tickers) now = datetime.now().strftime("%H:%M:%S") mode = " [가상]" if self.paper_mode else (" [복리]" if self.compound else "") lines = [f"[실시간 현황] {now}{mode}", ""] total_invested = 0 total_value = 0 for pos in holding: cur = prices.get(pos["ticker"]) if cur and pos["buy_price"] and pos["buy_volume"]: rate = ((cur - pos["buy_price"]) / pos["buy_price"]) * 100 pnl = (cur - pos["buy_price"]) * pos["buy_volume"] invested = pos["buy_price"] * pos["buy_volume"] total_invested += invested total_value += cur * pos["buy_volume"] strategy_name = pos.get("strategy_name") or get_strategy_profile(pos["ticker"])["label"] arrow = "+" if rate > 0 else "-" if rate < 0 else "=" lines.append( f" [{arrow}] {pos['coin']}: {rate:+.2f}% ({pnl:+,.0f}원)" f"\n {pos['buy_price']:,.0f} -> {cur:,.0f} | {strategy_name}" ) total_unrealized = total_value - total_invested lines.append("") lines.append(f"미실현: {total_unrealized:+,.0f}원") lines.append(f"실현: {self.total_realized:+,.0f}원") lines.append(f"총손익: {self.total_realized + total_unrealized:+,.0f}원") return "\n".join(lines) async def get_status_text(self) -> str: if not self.running: return "자동매매 비활성 상태\n/auto_start <투자금> 으로 시작" mode = " [가상]" if self.paper_mode else (" [복리]" if self.compound else "") budget_extra = f" (초기: {self.total_budget:,.0f}원)" if self.compound and self.total_realized != 0 else "" lines = [ f"[자동매매 현황]{mode}", f"시작: {self.started_at}", f"투자금: {self.effective_budget:,.0f}원{budget_extra}", f"전략: {self._strategy or '스캔 대기'} | 보유: {len(self.positions)}종목", f"누적수익: {self.total_realized:+,.0f}원", f"총거래: {self.total_trades}회", f"포지션: {len(self.positions)}/{self.max_coins}", "", ] if self._last_scan_at: stats = self._last_scan_stats trend_rejected = sum( stats.get(key, 0) for key in ("trend_down", "trend_neutral", "trend_up") ) lines.extend( [ f"최근 스캔: {self._last_scan_at.strftime('%m-%d %H:%M:%S')} | 시장기준 BTC {self._market_trend}", f"스캔 결과: 후보 {stats.get('candidates', 0)} / {stats.get('scanned', 0)}종목", ( f"필터 탈락: 추세/시장 {trend_rejected} | " f"데이터 {stats.get('daily_data', 0) + stats.get('signal_data', 0)} | " f"일봉ATR {stats.get('daily_atr', 0)} | " f"15분신호 {stats.get('signal_filter', 0)} | " f"수급 {stats.get('order_flow_blocked', 0) + stats.get('order_flow_data', 0)}" ), ] ) if self._last_scan_analyzed: lines.append( "15분 분석 종목: " + ", ".join(self._last_scan_analyzed) ) lines.append( "BTC는 시장 위험 기준이며 알트코인은 각각 독립 분석" ) if self._last_scan_candle is None: lines.append("다음 스캔: 즉시 실행 대기") else: now = datetime.now() minutes_to_next = 15 - (now.minute % 15) next_scan = (now + timedelta(minutes=minutes_to_next)).replace( second=0, microsecond=0 ) lines.append(f"다음 스캔: {next_scan.strftime('%m-%d %H:%M')} 전후") lines.append("") if self.positions: lines.append("[보유 포지션]") for pos in self.positions.values(): if pos["status"] == "holding" and pos["buy_price"]: try: cur = await asyncio.to_thread(pyupbit.get_current_price, pos["ticker"]) if cur: rate = ((cur - pos["buy_price"]) / pos["buy_price"]) * 100 pnl = (cur - pos["buy_price"]) * pos["buy_volume"] strategy_name = pos.get("strategy_name") or get_strategy_profile(pos["ticker"])["label"] lines.append( f" {pos['coin']}: {rate:+.2f}% ({pnl:+,.0f}원)\n" f" 매수: {pos['buy_price']:,.0f} → 현재: {cur:,.0f}\n" f" 전략: {strategy_name}\n" f" 목표: +{pos['target_profit']}% | 손절: -{pos['stop_loss']}%" ) except Exception: lines.append(f" {pos['coin']}: 조회 중...") else: lines.append(f" {pos['coin']}: {pos['status']}") unrealized = 0 for pos in self.positions.values(): if pos["status"] == "holding" and pos["buy_price"]: try: cur = await asyncio.to_thread(pyupbit.get_current_price, pos["ticker"]) if cur: unrealized += (cur - pos["buy_price"]) * pos["buy_volume"] except Exception: pass lines.append(f"\n미실현: {unrealized:+,.0f}원") lines.append(f"총손익: {self.total_realized + unrealized:+,.0f}원") if self._candidates: lines.append(f"\n대기 후보: {len(self._candidates)}개") return "\n".join(lines)