import pyupbit import pandas as pd import asyncio from upbit_trader import ( analyze_coin, check_sell_signal, check_trend, get_strategy_kind, get_strategy_profile, calc_atr_pct, get_entry_trend_adjustment, ) def _daily_history_before(df_daily: pd.DataFrame | None, current_time) -> pd.DataFrame | None: if df_daily is None: return None cutoff = current_time.replace(hour=9, minute=0, second=0, microsecond=0) rows = df_daily[df_daily.index < cutoff] if len(rows) < 60: return None return rows.tail(60).copy() async def run_backtest(ticker: str, days: int = 30, budget: float = 1000000) -> dict: """과거 데이터로 전략 백테스트 실행 Args: ticker: "KRW-BTC" 형식 days: 테스트 기간 (일) budget: 초기 투자금 Returns: 결과 딕셔너리 (승률, 수익률, 거래 내역 등) """ # 15분봉: 하루 96개 캔들 count = min(days * 96, 9600) # pyupbit 최대 제한 고려 df = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="minute15", count=count ) if df is None or len(df) < 100: return {"error": "데이터 부족"} # 일봉 데이터 (BTC 추세 필터 + 개별 코인 추세) df_daily = await asyncio.to_thread( pyupbit.get_ohlcv, ticker, interval="day", count=max(days + 60, 120) ) # BTC 일봉 (시장 필터) df_btc_daily = None if ticker != "KRW-BTC": df_btc_daily = await asyncio.to_thread( pyupbit.get_ohlcv, "KRW-BTC", interval="day", count=max(days + 60, 120) ) FEE_RATE = 0.0005 strategy_kind = get_strategy_kind(ticker) profile = get_strategy_profile(strategy_kind) trades = [] position = None equity = budget max_equity = budget max_drawdown = 0 win_count = 0 loss_count = 0 cooldown_until_index = 0 # 슬라이딩 윈도우로 시뮬레이션 window = 100 # analyze_coin에 필요한 최소 데이터 for i in range(window, len(df)): current_time = df.index[i] cur_price = df["close"].iloc[i] # 포지션 보유 중 → 매도 체크 if position: df_slice = df.iloc[i - window + 1:i + 1].copy() should_sell, reason = check_sell_signal( df_slice, position["buy_price"], position["target"], position["stop"], held_candles=i - position["buy_idx"], strategy_kind=strategy_kind, ) if should_sell: sell_price = cur_price gross = (sell_price - position["buy_price"]) * position["volume"] buy_fee = position["buy_price"] * position["volume"] * FEE_RATE sell_fee = sell_price * position["volume"] * FEE_RATE profit = gross - buy_fee - sell_fee profit_rate = ((sell_price - position["buy_price"]) / position["buy_price"]) * 100 equity += profit max_equity = max(max_equity, equity) dd = ((max_equity - equity) / max_equity) * 100 max_drawdown = max(max_drawdown, dd) if profit >= 0: win_count += 1 cooldown_until_index = i + profile["cooldown_win"] else: loss_count += 1 cooldown_until_index = i + profile["cooldown_loss"] trades.append({ "buy_time": position["buy_time"], "sell_time": current_time, "buy_price": position["buy_price"], "sell_price": sell_price, "profit_rate": round(profit_rate, 2), "profit_krw": round(profit, 0), "reason": reason, }) position = None continue # 포지션 없음 → 매수 체크 if i < cooldown_until_index: continue btc_trend = "neutral" # BTC 추세는 메이저 진입 조건과 알트 상대강도 점수에 반영한다. if df_btc_daily is not None and len(df_btc_daily) >= 50: btc_before = _daily_history_before(df_btc_daily, current_time) if btc_before is not None: btc_trend, _ = check_trend(btc_before) # 개별 코인은 일봉 상승 추세만 진입 대상으로 삼는다. trend_adjustment = None if df_daily is not None: daily_before = _daily_history_before(df_daily, current_time) if daily_before is not None: daily_atr_pct = calc_atr_pct(daily_before) if ( daily_atr_pct is None or daily_atr_pct < profile["min_daily_atr_pct"] or daily_atr_pct > profile["max_daily_atr_pct"] ): continue trend, _ = check_trend(daily_before) trend_adjustment = get_entry_trend_adjustment( ticker, trend, btc_trend ) if trend_adjustment is None: continue if trend_adjustment is None: continue df_slice = df.iloc[i - window + 1:i + 1].copy() result = analyze_coin(df_slice, ticker) if result: result["score"] += trend_adjustment if result and result["score"] >= profile["entry_threshold"]: invest = min(equity, budget) # 단일 종목 테스트 if invest < 5500: continue volume = invest / cur_price position = { "buy_price": cur_price, "volume": volume, "target": result["target_profit"], "stop": result["stop_loss"], "buy_time": current_time, "buy_idx": i, } # 미체결 포지션 강제 청산 if position: sell_price = df["close"].iloc[-1] gross = (sell_price - position["buy_price"]) * position["volume"] buy_fee = position["buy_price"] * position["volume"] * FEE_RATE sell_fee = sell_price * position["volume"] * FEE_RATE profit = gross - buy_fee - sell_fee profit_rate = ((sell_price - position["buy_price"]) / position["buy_price"]) * 100 equity += profit if profit >= 0: win_count += 1 else: loss_count += 1 trades.append({ "buy_time": position["buy_time"], "sell_time": df.index[-1], "buy_price": position["buy_price"], "sell_price": sell_price, "profit_rate": round(profit_rate, 2), "profit_krw": round(profit, 0), "reason": "강제청산(백테스트종료)", }) total_trades = win_count + loss_count total_return = ((equity - budget) / budget) * 100 win_rate = (win_count / total_trades * 100) if total_trades > 0 else 0 avg_profit = sum(t["profit_rate"] for t in trades if t["profit_rate"] > 0) / win_count if win_count > 0 else 0 avg_loss = sum(t["profit_rate"] for t in trades if t["profit_rate"] < 0) / loss_count if loss_count > 0 else 0 return { "ticker": ticker, "coin": ticker.replace("KRW-", ""), "strategy": profile["label"], "days": days, "budget": budget, "final_equity": round(equity, 0), "total_return": round(total_return, 2), "total_trades": total_trades, "win_count": win_count, "loss_count": loss_count, "win_rate": round(win_rate, 1), "avg_profit": round(avg_profit, 2), "avg_loss": round(avg_loss, 2), "max_drawdown": round(max_drawdown, 2), "trades": trades, } def format_backtest_result(result: dict) -> str: if "error" in result: return f"백테스트 실패: {result['error']}" lines = [ f"[백테스트 결과] {result['coin']}", f"━━━━━━━━━━━━━━━", f"전략: {result.get('strategy', '-')}", f"기간: {result['days']}일", f"초기자금: {result['budget']:,.0f}원", f"최종자금: {result['final_equity']:,.0f}원", f"총수익률: {result['total_return']:+.2f}%", f"━━━━━━━━━━━━━━━", f"총거래: {result['total_trades']}회", f"승: {result['win_count']}회 / 패: {result['loss_count']}회", f"승률: {result['win_rate']}%", f"평균익절: {result['avg_profit']:+.2f}%", f"평균손절: {result['avg_loss']:+.2f}%", f"최대낙폭: {result['max_drawdown']:.2f}%", ] # 최근 거래 5건 if result["trades"]: lines.append(f"\n[최근 거래]") for t in result["trades"][-5:]: emoji = "+" if t["profit_rate"] >= 0 else "-" lines.append( f" [{emoji}] {t['profit_rate']:+.2f}% ({t['profit_krw']:+,.0f}원)" f"\n {t['reason']}" ) return "\n".join(lines)