오늘은 가상 자산이 아닌 현실 자산을 이용해서 거래를하는 기능을 추가해 보겠습니다!!
드디어 메모장에 적힌 숫자가 아닌 계좌에 적힌 숫자로 거래를 하게 됩니다 ㅎㅎ
계획은 다음과 같습니다
- main_trader.py에 기존 매매 동작을 class로 정의!
- 세부 내용 살포시 변경!
- is_virtual이라는 변수를 통해 가상 거래를 할지, 현실 거래를 할지 선택!!
그럼 시작해 보겠습니다.
1. 기존 기능 class화
1). class TradingBot()
# ----------------- 클래스 정의 시작 -----------------
class TradingBot:
def __init__(self):
"""봇이 처음 생성될 때 실행되는 초기화 함수. 봇의 모든 상태 변수를 여기서 관리."""
print("🤖 트레이딩 봇 초기화 시작...")
# --- 봇 설정값 ---
self.analysis_interval = "30m"
self.check_interval_seconds = 30 * 60
self.settle_count = 5 # 중간 정산 기준 거래 횟수
self.consecutive_loss_limit = 5 # 연속 손실 한도
self.cooldown_hours = 6 # 거래 중지 시간 (hour)
# --- 봇 상태값 ---
self.vportfolio = virtual_portfolio_manager.load_vportfolio()
self.trade_stats = {"total_trades": 0, "wins": 0, "losses": 0, "total_gain": 0.0, "total_loss": 0.0}
self.buy_rate = 0.1
self.hodl_list = self.vportfolio.get("hodl_list", [])
# --- 초기 자산 계산 ---
self.initial_capital = self.vportfolio.get("cash", 1000000.0)
for coin in self.vportfolio.get('coins_owned', []):
self.initial_capital += coin["quantity"] * coin["avg_buy_price"]
trade_history.set_trade_mode(is_virtual=True)
# --- 봇 시작 메시지 ---
print("=" * 50)
print(f"🚀 가상 자동매매 봇을 시작합니다! (HODL 리스트: {self.hodl_list})")
print(f"⏳ 분석 간격: {self.analysis_interval}, 체크 주기: {self.check_interval_seconds}초")
print(f"💰 시작 총 자산 : {self.initial_capital:,.0f} KRW")
print("봇을 중지하려면 Ctrl+C 를 누르세요.")
print("=" * 50)
def _filter_top_coins(self, top_n=20):
"""거래대금 상위 코인을 필터링하는 내부 메소드"""
# 1. API를 통해 전체 코인 Ticker 정보를 가져옴
# get_ticker_info 함수는 성공 시 'data' 딕셔너리를 반환함
all_ticker_data = bithumb_api_client.get_ticker_info("ALL")
# 2. API 호출 실패 또는 데이터 형식 오류 확인
if not all_ticker_data or not isinstance(all_ticker_data, dict):
print("⚠️ 전체 Ticker 정보 조회에 실패하여 필터링을 건너뜁니다.")
return []
# 3. 데이터 처리
coin_tickers = all_ticker_data.copy() # 원본 수정을 피하기 위해 복사
if 'date' in coin_tickers:
del coin_tickers['date']
# 거래대금을 기준으로 코인들을 정렬하기 위해 리스트로 변환
sorted_coins = []
for symbol, data in coin_tickers.items():
try:
sorted_coins.append({'symbol': symbol, 'value': float(data.get('acc_trade_value_24H', 0))})
except (ValueError, TypeError):
continue
# 거래대금을 기준으로 내림차순 정렬
sorted_coins.sort(key=lambda x: x['value'], reverse=True)
# 거래대금 상위 N개 코인 데이터 추출
top_coins_data = sorted_coins[:top_n]
target_coins = [coin['symbol'] for coin in top_coins_data]
'''
print("\n--- 💰 24시간 거래대금 상위 20개 관심 종목 ---")
# 리스트가 비어있지 않을 때만 출력
if top_coins_data:
print("거래대금 상위 코인을 찾지 못했습니다.")
for i, coin in enumerate(top_coins_data):
print(f"{i+1:>2}. {coin['symbol']:<10} (거래대금: {coin['value']:>15,.0f} KRW)")
else:
print("거래대금 상위 코인을 찾지 못했습니다.")
'''
return target_coins
def _check_cross_signal(self, df):
"""골든/데드 크로스를 판단하는 내부 메소드"""
last_row, prev_row = df.iloc[-1], df.iloc[-2]
sma5_today, sma20_today = last_row.get('SMA5'), last_row.get('SMA20')
sma5_yesterday, sma20_yesterday = prev_row.get('SMA5'), prev_row.get('SMA20')
if None in [sma5_today, sma20_today, sma5_yesterday, sma20_yesterday]: return "HOLD"
if sma5_yesterday < sma20_yesterday and sma5_today > sma20_today:
return "BUY"
elif sma5_yesterday > sma20_yesterday and sma5_today < sma20_today:
return "SELL"
else:
return "HOLD"
def _check_and_execute_sell_strategy(self, coin_symbol, owned_coin_info):
"""보유 코인에 대한 매도 전략을 확인하고 실행합니다."""
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if not ticker:
print(f"-> [{coin_symbol}] 현재가 조회 실패. 매도 판단을 건너뜁니다.")
return
current_price = float(ticker['closing_price'])
avg_buy_price = owned_coin_info['avg_buy_price']
profit_rate = (current_price - avg_buy_price) / avg_buy_price if avg_buy_price > 0 else 0
pt_rate = self.vportfolio['strategy_params']['profit_take_percentage']
sl_rate = self.vportfolio['strategy_params']['stop_loss_percentage']
signal = "HOLD"
candlestick_data = bithumb_api_client.get_candlestick_data(coin_symbol, chart_intervals=self.analysis_interval)
if candlestick_data and len(candlestick_data) > 20:
df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
df['close'] = pd.to_numeric(df['close'])
df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)
signal = self._check_cross_signal(df)
print(f"-> [{coin_symbol}] 보유 중. 기술적 신호: {signal} (현재 수익률: {profit_rate:+.2%}) ")
is_profit_take = profit_rate >= pt_rate
is_stop_loss = profit_rate <= -sl_rate
is_dead_cross = signal == "SELL"
should_sell = False
sell_reason = ""
quantity_to_sell = 0
if is_profit_take:
sell_reason = "profit_take"
quantity_to_sell = owned_coin_info['quantity'] # 전량 매도
print(f"\t-> [{coin_symbol}] 💰 수익실현 조건 달성! 전량 매도를 시도합니다.")
should_sell = True
elif is_stop_loss:
sell_reason = "stop_loss"
quantity_to_sell = owned_coin_info['quantity'] # 전량 매도
print(f"\t-> [{coin_symbol}] 🛡️ 손절매 조건 달성! 전량 매도를 시도합니다.")
should_sell = True
elif is_dead_cross:
# 데드크로스는 추세 전환 신호이므로, 일단 절반만 팔아서 리스크를 줄이는 아이디어 적용!
sell_reason = "dead_cross"
quantity_to_sell = owned_coin_info['quantity'] / 2 # 50% 매도
print(f"\t-> [{coin_symbol}] 🔴 데드 크로스 발생! 50% 매도를 시도합니다.")
should_sell = True
# --- 3. 매도 계획이 세워졌다면, 최종적으로 딱 한 번만 실행한다 ---
if should_sell:
profit = (current_price - avg_buy_price) * quantity_to_sell
# --- 👇 연속 손실 횟수 업데이트 로직 추가! 👇 ---
if profit >= 0: # 수익이거나 본전일 때
# 수익이 나면 연속 손실 횟수를 0으로 리셋
self.vportfolio["consecutive_losses"] = 0
print(f"-> 👍 수익 발생. 연속 손실 횟수가 0으로 초기화됩니다.")
else: # 손실일 때
# 손실이 나면 연속 손실 횟수를 1 증가
self.vportfolio["consecutive_losses"] += 1
print(f"-> 👎 손실 발생. 연속 손실 횟수 +1 (현재: {self.vportfolio['consecutive_losses']}회)")
# --- 연속 손실 한도 도달 시 쿨다운 설정! ---
if self.vportfolio["consecutive_losses"] >= self.consecutive_loss_limit:
cooldown_end_time = time.time() + self.cooldown_hours * 3600 # hour
self.vportfolio["cooldown_until"] = cooldown_end_time
print(f"🔥🔥🔥 연속 손실 한도({self.consecutive_loss_limit}회) 도달! {self.cooldown_hours}시간 동안 거래를 중지합니다.")
print(f"-> 거래 재개 예정 시각: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(cooldown_end_time))}")
# 거래 통계 업데이트
if sell_reason == "profit_take":
self.trade_stats["wins"] += 1; self.trade_stats["total_gain"] += profit
elif sell_reason == "stop_loss":
self.trade_stats["losses"] += 1; self.trade_stats["total_loss"] += profit
elif sell_reason == "dead_cross":
if profit > 0:
self.trade_stats["wins"] += 1; self.trade_stats["total_gain"] += profit
else:
self.trade_stats["losses"] += 1; self.trade_stats["total_loss"] += profit
self.trade_stats["total_trades"] += 1
# 가상 매도 실행
updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(
self.vportfolio, "sell", coin_symbol, quantity_to_sell, current_price,
reason=sell_reason, profit_rate=profit_rate
)
if updated_portfolio: self.vportfolio = updated_portfolio
def _check_and_execute_buy_strategy(self, coin_symbol):
"""미보유 코인에 대한 매수 전략을 확인하고 실행합니다."""
candlestick_data = bithumb_api_client.get_candlestick_data(coin_symbol, chart_intervals=self.analysis_interval)
signal = "HOLD"
if candlestick_data and len(candlestick_data) > 20:
df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
df['close'] = pd.to_numeric(df['close'])
df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)
signal = self._check_cross_signal(df)
print(f"-> [{coin_symbol}] 미보유. 기술적 신호: {signal}")
if signal == "BUY":
print(f"\t-> [{coin_symbol}] 🟢 골든 크로스 발생! 신규 매수를 시도합니다.")
cash_to_invest = self.vportfolio['cash'] * self.buy_rate
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if ticker:
current_price = float(ticker['closing_price'])
quantity_to_buy = cash_to_invest / current_price if current_price > 0 else 0
updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(
self.vportfolio, "buy", coin_symbol, quantity_to_buy, current_price
)
if updated_portfolio: self.vportfolio = updated_portfolio
def _check_and_settle(self):
"""중간 정산 로직"""
if self.trade_stats["total_trades"] >= self.settle_count:
odds = self.trade_stats["wins"] / self.trade_stats["total_trades"]
print("=" * 50)
print(f"⭐[중간 정산] 최근 {self.trade_stats['total_trades']}회 거래 실적⭐")
print(f"승리 횟수: {self.trade_stats['wins']} / 총 수익 금액: {self.trade_stats['total_gain']:,.0f} KRW")
print(f"패배 횟수: {self.trade_stats['losses']} / 총 손실 금액: {self.trade_stats['total_loss']:,.0f} KRW")
print(f"승률 : {odds:.2%}")
# 승률에 따른 매수 비율 조정
if odds >= 0.6 and self.buy_rate < 0.5:
old_buy_rate = self.buy_rate
self.buy_rate += 0.1
print(f" --> 높은 승률로 매수 비율이 상승🔼하였습니다. [ {old_buy_rate:.1%} --> {self.buy_rate:.1%}]")
elif odds <= 0.4 and self.buy_rate > 0.1:
old_buy_rate = self.buy_rate
self.buy_rate -= 0.1
print(f" --> 낮은 승률로 매수 비율이 하락🔽하였습니다. [ {old_buy_rate:.1%} --> {self.buy_rate:.1%}]")
print("=" * 50)
# 정산 후 통계 초기화
self.trade_stats = {"total_trades": 0, "wins": 0, "losses": 0, "total_gain": 0.0, "total_loss": 0.0}
def _print_total_assets(self):
"""총 자산 현황 출력 로직"""
print("\n" + "=" * 20 + " ⚖️ 포트폴리오 현황 ⚖️ " + "=" * 20)
total_coin_value = 0.0
for coin in self.vportfolio.get('coins_owned', []):
ticker = bithumb_api_client.get_ticker_info(coin["symbol"])
if ticker:
current_price = float(ticker['closing_price'])
coin_value = coin["quantity"] * current_price
total_coin_value += coin_value
if coin['avg_buy_price'] < 10.0 or current_price < 10.0:
print(f" - 보유: {coin['symbol']:<8} 수량: {coin['quantity']:<15.4f} 평단가: {coin['avg_buy_price']:>12,.4f} KRW | 현재가: {current_price:>12,.4f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(coin['avg_buy_price']-current_price)/coin['avg_buy_price']:>6,.1%}")
else:
print(f" - 보유: {coin['symbol']:<8} 수량: {coin['quantity']:<15.4f} 평단가: {coin['avg_buy_price']:>12,.0f} KRW | 현재가: {current_price:>12,.0f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(coin['avg_buy_price']-current_price)/coin['avg_buy_price']:>6,.1%}")
current_cash = self.vportfolio.get("cash", 0.0)
total_assets = current_cash + total_coin_value
pnl_percentage = ((total_assets - self.initial_capital) / self.initial_capital) * 100 if self.initial_capital > 0 else 0
print(f" - 현금: {current_cash:,.0f} KRW")
print("-" * 62)
print(f" - 총 자산: {total_assets:,.0f} KRW (초기 자산 대비: {pnl_percentage:+.2f}%)")
print("=" * 62)
def run_one_cycle(self):
"""한 번의 분석/거래 사이클 전체를 지휘하는 메인 메소드"""
# --- 쿨다운(거래 중지) 상태인지 먼저 확인! ---
is_in_cooldown = False
cooldown_until = self.vportfolio.get("cooldown_until", 0)
if time.time() < cooldown_until:
is_in_cooldown = True
resume_time_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(cooldown_until))
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ❄️ 거래 중지(쿨다운) 상태입니다. ({resume_time_str} 까지)")
elif cooldown_until > 0:
# 쿨다운 시간이 지났다면, 상태를 초기화
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ✅ 쿨다운 기간이 종료되었습니다. 거래를 재개합니다.")
self.vportfolio["consecutive_losses"] = 0
self.vportfolio["cooldown_until"] = 0
virtual_portfolio_manager.save_vportfolio(self.vportfolio) # 상태 초기화 후 저장
# 1. 분석 대상 선정
top_coins = self._filter_top_coins()
owned_symbols = [coin['symbol'] for coin in self.vportfolio.get('coins_owned', [])]
combined_list = list(set(top_coins + owned_symbols))
final_analysis_list = [coin for coin in combined_list if coin not in self.hodl_list]
print(f"\n--- 최종 분석 대상 ({len(final_analysis_list)}개) ---")
print(final_analysis_list)
print("-" * 50)
# 2. 각 코인 분석 및 거래 실행
for coin_symbol in final_analysis_list:
owned_coin_info = next((coin for coin in self.vportfolio["coins_owned"] if coin["symbol"] == coin_symbol), None)
if owned_coin_info:
self._check_and_execute_sell_strategy(coin_symbol, owned_coin_info)
else:
# 쿨다운 중에는 매수 중지
if is_in_cooldown:
continue
self._check_and_execute_buy_strategy(coin_symbol)
# 3. 사이클 종료 후 중간 정산 및 자산 현황 출력
self._check_and_settle()
self._print_total_assets()2). main 변경
# --- 메인 실행 부분 ---
if __name__ == "__main__":
my_bot = TradingBot()
while True: # 실제 운영 시 주석 해제
try:
my_bot.run_one_cycle()
time.sleep(my_bot.check_interval_seconds) # 실제 운영 시 주석 해제
except KeyboardInterrupt:
print("\n\n사용자에 의해 프로그램이 중단되었습니다. 최종 포트폴리오 상태를 저장합니다...")
virtual_portfolio_manager.save_vportfolio(my_bot.vportfolio)
print("프로그램을 종료합니다.")
break
except Exception as e:
print(f"⚠️ 예상치 못한 오류 발생: {e}")
print(f"{my_bot.check_interval_seconds}초 후 재시도합니다...")
time.sleep(my_bot.check_interval_seconds)
2. 현실 자산 거래 함수 추가
현실 자산 거래를 위해 bithumb_api_client.py에 다음 함수를 추가해줍니다.
1). import 추가
import hashlib
from urllib.parse import urlencode2). 헤더 생성 함수
# --- Private API 호출을 위한 헬퍼(Helper) 함수 ---
"""
Private API 요청을 위한 인증 헤더(JWT)를 생성
:endpoint: 요청할 엔드포인트 (예: "/v1/accounts")
:params: POST 요청 시 body에 포함될 파라미터 딕셔너리
:return: 인증 헤더 딕셔너리
"""
def _get_v1_auth_headers(params_for_hash={}):
"""v1 Private API 요청을 위한 JWT 인증 헤더를 생성합니다."""
payload = {
'access_key': api_key,
'nonce': str(uuid.uuid4()),
'timestamp': str(round(time.time() * 1000)),
}
if params_for_hash:
query_string = urlencode(params_for_hash)
hasher = hashlib.sha512()
hasher.update(query_string.encode('utf-8'))
query_hash = hasher.hexdigest()
payload['query_hash'] = query_hash
payload['query_hash_alg'] = 'SHA512'
jwt_token = jwt.encode(payload, api_secret, algorithm="HS256")
return {
'Authorization': f'Bearer {jwt_token}',
'Content-Type': 'application/json'
}3). 실제 매수/매도 함수
전체 매도를 진행하면 수수료 에러가 발생하는 경우가 있어서 dust를 남기게 되지만 (개수 / 1.0025)만큼만 팔 수 있게 설정했어요.
그리고 최소 주문 금액이 5천원 인것도 꼭 기억해야 합니다!
이거 계속 테스트하느라 수수료 계속 날아가 버렸네요ㅠㅠ
"""
v1 API로 실제 매수/매도 주문 (Private API)
:side: 'buy' 또는 'sell'
:ord_type: 'limit', 'market_buy', 'market_sell'
:units: 주문 수량 (코인 개수)
:price: 주문 가격 또는 총액 (KRW)
"""
def place_order(order_currency, side, ord_type, units=None, price=None, payment_currency="KRW"):
if not api_key or not api_secret: return None
endpoint = "/v1/orders"
url = BITHUMB_API_URL + endpoint
request_body = {
"market": f"{payment_currency.upper()}-{order_currency.upper()}",
"side": "bid" if side.lower() == "buy" else "ask",
}
# 수수료때문에 매도가 되지 않는 문제 해결
if side.lower() == "sell" and units is not None:
fee_rate = 0.0025
units = units / (1.0 + fee_rate)
ord_type_lower = ord_type.lower()
if ord_type_lower == 'limit':
if units is None or price is None: print("🔥 지정가 주문에는 units와 price가 모두 필요합니다."); return None
request_body['ord_type'] = 'limit';
request_body['volume'] = f"{units:.4f}";
request_body['price'] = str(price)
elif ord_type_lower == 'market_buy':
if price is None: print("🔥 시장가 매수에는 price(총 주문금액)가 필요합니다."); return None
request_body['ord_type'] = 'price';
request_body['price'] = str(price)
elif ord_type_lower == 'market_sell':
if units is None: print("🔥 시장가 매도에는 units(수량)가 필요합니다."); return None
request_body['ord_type'] = 'market';
request_body['volume'] = f"{units:.4f}"
else:
print(f"🔥 지원하지 않는 주문 유형입니다: {ord_type}");
return None
try:
headers = _get_v1_auth_headers(request_body)
response = requests.post(url, headers=headers, data=json.dumps(request_body))
response.raise_for_status()
result = response.json()
if isinstance(result, dict) and 'uuid' in result:
order_id = result.get('uuid')
side_kor = "매수" if request_body["side"] == "bid" else "매도"
print(f"✅ [실제 주문 성공] {order_currency} {side_kor} 주문이 접수되었습니다. (주문 ID: {order_id})")
return result
else:
# 실패 응답은 보통 'code'와 'message'를 포함함
print(f"🔥 [실제 주문 실패] API 에러: {result.get('message')} (코드: {result.get('code')})")
return None
except requests.exceptions.HTTPError as e:
print(f"주문 중 에러 발생: {e.response.status_code} - {e.response.text}")
return None
except Exception as e:
print(f"주문 중 에러 발생: {e}")
return None4). 가상 거래에 수수료 추가
겸사겸사 가상 거래에도 수수료를 추가해줍니다.
virtual_portfolio_manager.py의 record_vportfolio_trade함수 변경
def record_vportfolio_trade(portfolio_data, trade_type, symbol, quantity, price_per_unit, reason=None, profit_rate=0.0):
symbol_upper = symbol.upper()
total_value = quantity * price_per_unit
# --- 👇 현실적인 수수료 계산 로직 추가! 👇 ---
fee_rate = 0.0025 # 빗썸 수수료율 0.25%
trade_fee = total_value * fee_rate
# --- 👆 여기까지 👆 ---
trade_status_message = ""
if trade_type.lower() == "buy":
# --- 👇 매수 시, (거래대금 + 수수료) 만큼 현금 차감 👇 ---
required_cash = total_value + trade_fee
if portfolio_data["cash"] < required_cash:
print(f"🔥 [가상 매수 실패] 현금 부족. (필요: {required_cash:,.0f} KRW, 보유: {portfolio_data['cash']:,.0f} KRW)")
return None # 거래 실패
portfolio_data["cash"] -= required_cash
# --- 👆 여기까지 👆 ---
# 보유 코인 목록 업데이트
owned = False
for coin in portfolio_data["coins_owned"]:
if coin["symbol"] == symbol_upper:
# 이미 보유한 코인이면, 평단가 재계산 (이동평균법)
current_total_value = coin["quantity"] * coin["avg_buy_price"]
new_total_value = current_total_value + total_value
new_total_quantity = coin["quantity"] + quantity
coin["avg_buy_price"] = new_total_value / new_total_quantity
coin["quantity"] = new_total_quantity
owned = True
break
if not owned:
portfolio_data["coins_owned"].append({
"symbol": symbol_upper, "quantity": quantity, "avg_buy_price": price_per_unit
})
trade_status_message = f"✅ [가상 매수 성공] {symbol_upper} {quantity:.4f}개 매수. (수수료: {trade_fee:,.2f} KRW 차감)"
elif trade_type.lower() == "sell":
owned_coin = next((c for c in portfolio_data["coins_owned"] if c["symbol"] == symbol_upper), None)
if not owned_coin or owned_coin["quantity"] < quantity:
print(
f"🔥 [가상 매도 실패] 보유 수량 부족. (판매 시도: {quantity:.4f}, 보유: {owned_coin['quantity'] if owned_coin else 0:.4f})")
return None
# --- 👇 매도 시, 거래대금에서 수수료를 뺀 만큼만 현금 증가 👇 ---
portfolio_data["cash"] += (total_value - trade_fee)
# --- 👆 여기까지 👆 ---
owned_coin["quantity"] -= quantity
if owned_coin["quantity"] <= 1e-8: # 아주 작은 수량은 0으로 처리하고 목록에서 제거
portfolio_data["coins_owned"] = [c for c in portfolio_data["coins_owned"] if c["symbol"] != symbol_upper]
trade_status_message = f"✅ [가상 매도 성공] {symbol_upper} {quantity:.4f}개 매도. (수수료: {trade_fee:,.2f} KRW 차감)"
print(trade_status_message)
# 거래 내역 생성 및 기록
trade_info = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "type": trade_type.lower(), "symbol": symbol_upper,
"quantity": quantity, "price_per_unit": price_per_unit, "total_value": total_value,
"fee": trade_fee, "cash_after_trade": portfolio_data["cash"],
"reason": reason, "profit_rate": profit_rate
}
trade_history.add_trade_to_history(trade_info)
save_vportfolio(portfolio_data)
return portfolio_data
3. main_trader.py 변경
main_trader.py는 바꿀게 정말 많아요!
그래서 작업하다가 어디어디 작업했는지 까먹어버렸습니다!!ㅠㅠ
class 통소스 올릴테니 공부하실분은 열심히 분석해주세요
class TradingBot:
def __init__(self, is_virtual = True):
"""봇이 처음 생성될 때 실행되는 초기화 함수. 봇의 모든 상태 변수를 여기서 관리."""
print("🤖 트레이딩 봇 초기화 시작...")
# --- 봇 설정값 ---
self.is_virtual_mode = is_virtual
self.analysis_interval = "30m"
self.check_interval_seconds = 30 * 60
self.settle_count = 5 # 중간 정산 기준 거래 횟수
self.consecutive_loss_limit = 5 # 연속 손실 한도
self.cooldown_hours = 6 # 거래 중지 시간 (hour)
# --- 봇 상태값 ---
self.vportfolio = virtual_portfolio_manager.load_vportfolio()
self.trade_stats = {"total_trades": 0, "wins": 0, "losses": 0, "total_gain": 0.0, "total_loss": 0.0}
self.buy_rate = 0.1
self.hodl_list = self.vportfolio.get("hodl_list", [])
# --- 초기 자산 계산 ---
self.initial_capital = 0.0
self._initialize_capital()
trade_history.set_trade_mode(is_virtual=self.is_virtual_mode)
# --- 봇 시작 메시지 ---
print("=" * 50)
print(f"🚀 가상 자동매매 봇을 시작합니다! (HODL 리스트: {self.hodl_list})")
print(f"⏳ 분석 간격: {self.analysis_interval}, 체크 주기: {self.check_interval_seconds}초")
print(f"💰 시작 총 자산 : {self.initial_capital:,.0f} KRW")
print("봇을 중지하려면 Ctrl+C 를 누르세요.")
print("=" * 50)
def _initialize_capital(self):
if self.is_virtual_mode:
print("⚖ 가상 포트폴리오를 기준으로 초기 자산을 계산합니다 ⚖")
self.initial_capital = self.vportfolio.get("cash", 1000000.0)
for coin in self.vportfolio.get('coins_owned', []):
self.initial_capital += coin["quantity"] * coin["avg_buy_price"]
else:
print("⚖ 실제 계좌 정보을 기준으로 초기 자산을 계산합니다 ⚖")
real_account_info = bithumb_api_client.get_account_info()
if isinstance(real_account_info, list):
for asset in real_account_info:
currency = asset.get('currency')
balance = float(asset.get('balance', 0))
if currency == 'KRW':
self.initial_capital += balance
else:
# 코인은 평단가 * 수량으로 가치 계산
avg_buy_price = float(asset.get('avg_buy_price', 0))
self.initial_capital += balance * avg_buy_price
else:
print("🔥 실제 계좌 정보를 가져올 수 없어 초기 자산을 0으로 설정합니다.")
def _filter_top_coins(self, top_n=20):
"""거래대금 상위 코인을 필터링하는 내부 메소드"""
# 1. API를 통해 전체 코인 Ticker 정보를 가져옴
# get_ticker_info 함수는 성공 시 'data' 딕셔너리를 반환함
all_ticker_data = bithumb_api_client.get_ticker_info("ALL")
# 2. API 호출 실패 또는 데이터 형식 오류 확인
if not all_ticker_data or not isinstance(all_ticker_data, dict):
print("⚠️ 전체 Ticker 정보 조회에 실패하여 필터링을 건너뜁니다.")
return []
# 3. 데이터 처리
coin_tickers = all_ticker_data.copy() # 원본 수정을 피하기 위해 복사
if 'date' in coin_tickers:
del coin_tickers['date']
# 거래대금을 기준으로 코인들을 정렬하기 위해 리스트로 변환
sorted_coins = []
for symbol, data in coin_tickers.items():
try:
sorted_coins.append({'symbol': symbol, 'value': float(data.get('acc_trade_value_24H', 0))})
except (ValueError, TypeError):
continue
# 거래대금을 기준으로 내림차순 정렬
sorted_coins.sort(key=lambda x: x['value'], reverse=True)
# 거래대금 상위 N개 코인 데이터 추출
top_coins_data = sorted_coins[:top_n]
target_coins = [coin['symbol'] for coin in top_coins_data]
'''
print("\n--- 💰 24시간 거래대금 상위 20개 관심 종목 ---")
# 리스트가 비어있지 않을 때만 출력
if top_coins_data:
print("거래대금 상위 코인을 찾지 못했습니다.")
for i, coin in enumerate(top_coins_data):
print(f"{i+1:>2}. {coin['symbol']:<10} (거래대금: {coin['value']:>15,.0f} KRW)")
else:
print("거래대금 상위 코인을 찾지 못했습니다.")
'''
return target_coins
def _check_cross_signal(self, df):
"""골든/데드 크로스를 판단하는 내부 메소드"""
last_row, prev_row = df.iloc[-1], df.iloc[-2]
sma5_today, sma20_today = last_row.get('SMA5'), last_row.get('SMA20')
sma5_yesterday, sma20_yesterday = prev_row.get('SMA5'), prev_row.get('SMA20')
if None in [sma5_today, sma20_today, sma5_yesterday, sma20_yesterday]: return "HOLD"
if sma5_yesterday < sma20_yesterday and sma5_today > sma20_today:
return "BUY"
elif sma5_yesterday > sma20_yesterday and sma5_today < sma20_today:
return "SELL"
else:
return "HOLD"
def _check_and_execute_sell_strategy(self, coin_symbol, owned_coin_info):
"""보유 코인에 대한 매도 전략을 확인하고 실행합니다."""
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if not ticker:
print(f"-> [{coin_symbol}] 현재가 조회 실패. 매도 판단을 건너뜁니다.")
return
current_price = float(ticker['closing_price'])
avg_buy_price = owned_coin_info['avg_buy_price']
profit_rate = (current_price - avg_buy_price) / avg_buy_price if avg_buy_price > 0 else 0
pt_rate = self.vportfolio['strategy_params']['profit_take_percentage']
sl_rate = self.vportfolio['strategy_params']['stop_loss_percentage']
signal = "HOLD"
candlestick_data = bithumb_api_client.get_candlestick_data(coin_symbol, chart_intervals=self.analysis_interval)
if candlestick_data and len(candlestick_data) > 20:
df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
df['close'] = pd.to_numeric(df['close'])
df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)
signal = self._check_cross_signal(df)
print(f"-> [{coin_symbol}] 보유 중. 기술적 신호: {signal} (현재 수익률: {profit_rate:+.2%}) ")
is_profit_take = profit_rate >= pt_rate
is_stop_loss = profit_rate <= -sl_rate
is_dead_cross = signal == "SELL"
should_sell = False
sell_reason = ""
quantity_to_sell = 0
if is_profit_take:
sell_reason = "profit_take"
quantity_to_sell = owned_coin_info['quantity'] # 전량 매도
print(f"\t-> [{coin_symbol}] 💰 수익실현 조건 달성! 전량 매도를 시도합니다.")
should_sell = True
elif is_stop_loss:
sell_reason = "stop_loss"
quantity_to_sell = owned_coin_info['quantity'] # 전량 매도
print(f"\t-> [{coin_symbol}] 🛡️ 손절매 조건 달성! 전량 매도를 시도합니다.")
should_sell = True
elif is_dead_cross:
# 데드크로스는 추세 전환 신호이므로, 일단 절반만 팔아서 리스크를 줄이는 아이디어 적용!
sell_reason = "dead_cross"
quantity_to_sell = owned_coin_info['quantity'] / 2 # 50% 매도
print(f"\t-> [{coin_symbol}] 🔴 데드 크로스 발생! 50% 매도를 시도합니다.")
should_sell = True
# --- 3. 매도 계획이 세워졌다면, 최종적으로 딱 한 번만 실행한다 ---
if should_sell:
profit = (current_price - avg_buy_price) * quantity_to_sell
# --- 👇 연속 손실 횟수 업데이트 로직 추가! 👇 ---
if profit >= 0: # 수익이거나 본전일 때
# 수익이 나면 연속 손실 횟수를 0으로 리셋
self.vportfolio["consecutive_losses"] = 0
print(f"-> 👍 수익 발생. 연속 손실 횟수가 0으로 초기화됩니다.")
else: # 손실일 때
# 손실이 나면 연속 손실 횟수를 1 증가
self.vportfolio["consecutive_losses"] += 1
print(f"-> 👎 손실 발생. 연속 손실 횟수 +1 (현재: {self.vportfolio['consecutive_losses']}회)")
# --- 연속 손실 한도 도달 시 쿨다운 설정! ---
if self.vportfolio["consecutive_losses"] >= self.consecutive_loss_limit:
cooldown_end_time = time.time() + self.cooldown_hours * 3600 # hour
self.vportfolio["cooldown_until"] = cooldown_end_time
print(f"🔥🔥🔥 연속 손실 한도({self.consecutive_loss_limit}회) 도달! {self.cooldown_hours}시간 동안 거래를 중지합니다.")
print(f"-> 거래 재개 예정 시각: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(cooldown_end_time))}")
# 거래 통계 업데이트
if sell_reason == "profit_take":
self.trade_stats["wins"] += 1; self.trade_stats["total_gain"] += profit
elif sell_reason == "stop_loss":
self.trade_stats["losses"] += 1; self.trade_stats["total_loss"] += profit
elif sell_reason == "dead_cross":
if profit > 0:
self.trade_stats["wins"] += 1; self.trade_stats["total_gain"] += profit
else:
self.trade_stats["losses"] += 1; self.trade_stats["total_loss"] += profit
self.trade_stats["total_trades"] += 1
# 가상 매도 실행
updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(
self.vportfolio, "sell", coin_symbol, quantity_to_sell, current_price,
reason=sell_reason, profit_rate=profit_rate
)
if updated_portfolio: self.vportfolio = updated_portfolio
def _check_and_execute_buy_strategy(self, coin_symbol):
"""미보유 코인에 대한 매수 전략을 확인하고 실행합니다."""
candlestick_data = bithumb_api_client.get_candlestick_data(coin_symbol, chart_intervals=self.analysis_interval)
signal = "HOLD"
if candlestick_data and len(candlestick_data) > 20:
df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
df['close'] = pd.to_numeric(df['close'])
df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)
signal = self._check_cross_signal(df)
print(f"-> [{coin_symbol}] 미보유. 기술적 신호: {signal}")
if signal == "BUY":
print(f"\t-> [{coin_symbol}] 🟢 골든 크로스 발생! 신규 매수를 시도합니다.")
available_cash = self.vportfolio['cash']
min_order_krw = 5000 # 최소 주문 금액
fee_rate = 0.0025 # 수수료율
cash_to_invest = 0
if available_cash < min_order_krw:
print(f"\t-> 보유 현금({available_cash:,.0f}원)이 최소 주문 금액({min_order_krw:,.0f}원)보다 적어 매수를 건너뜁니다.")
return
# 보유 현금이 10,000원 이하면 전액 매수 시도
if available_cash <= 10000:
print(f"\t-> 보유 현금이 적어 전액 매수를 시도합니다.")
cash_to_invest = available_cash
else:
# 기본적으로는 설정된 비율(buy_rate)만큼 투자
intended_investment = available_cash * self.buy_rate
# 다만, 계산된 금액이 최소 주문 금액보다 작으면, 최소 주문 금액으로 투자
if intended_investment < min_order_krw:
print(f"\t-> 비율 투자금({intended_investment:,.0f}원)이 최소 금액보다 작아, 최소 금액({min_order_krw:,.0f}원)으로 투자를 시도합니다.")
cash_to_invest = min_order_krw
else:
cash_to_invest = intended_investment
# --- 4. 최종 투자 금액으로 실제 주문 ---
# 수수료를 고려하여, 실제 코인을 사는 데 사용할 순수 금액 계산
net_investment = cash_to_invest / (1 + fee_rate)
# 현재가 조회
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if ticker:
current_price = float(ticker['closing_price'])
if current_price <= 0: return # 가격이 0 이하면 주문 불가
quantity_to_buy = net_investment / current_price
updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(
self.vportfolio, "buy", coin_symbol, quantity_to_buy, current_price
)
if updated_portfolio: self.vportfolio = updated_portfolio
def _check_and_execute_real_sell_strategy(self, coin_symbol, owned_coin_info):
"""[실전 모드] 보유 코인에 대한 매도 전략을 확인하고 실제 주문을 실행합니다."""
print(f"-> [{coin_symbol}] 보유 중. 실전 매도 조건을 확인합니다.")
# --- 👇 핵심 수정 1: 'available' 수량 직접 계산! 👇 ---
balance = float(owned_coin_info.get('balance', 0))
locked = float(owned_coin_info.get('locked', 0))
available_quantity = balance - locked
if available_quantity <= 0:
print(f"\t-> [{coin_symbol}] 주문에 묶여있거나, 사용 가능한 수량이 없습니다.")
return
# --- 👆 여기까지 👆 ---
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if not ticker:
print(f"\t-> [{coin_symbol}] 현재가 조회 실패. 매도 판단을 건너뜁니다.")
return
current_price = float(ticker['closing_price'])
avg_buy_price = float(owned_coin_info.get('avg_buy_price', 0))
if avg_buy_price == 0:
print(f"\t-> [{coin_symbol}] 평단가 정보가 없어 수익률을 계산할 수 없습니다.")
return
profit_rate = (current_price - avg_buy_price) / avg_buy_price
print(f"\t-> [{coin_symbol}] 현재 수익률: {profit_rate:+.2%}")
# --- 👇 핵심 수정 2: vportfolio가 아닌 self에서 전략 파라미터 가져오기 👇 ---
pt_rate = self.vportfolio['strategy_params']['profit_take_percentage']
sl_rate = self.vportfolio['strategy_params']['stop_loss_percentage']
# --- 👆 (이 로직은 그대로 두는 것이 좋다고 판단!) 👆 ---
should_sell = False
if profit_rate >= pt_rate:
print(f"\t-> [{coin_symbol}] 💰 수익실현 조건 달성! 실제 매도를 시도합니다.")
should_sell = True
elif profit_rate <= -sl_rate:
print(f"\t-> [{coin_symbol}] 🛡️ 손절매 조건 달성! 실제 매도를 시도합니다.")
should_sell = True
if should_sell:
# 수수료 및 최소 주문 금액 문제를 고려하여, 실제 주문 가능한 수량 계산
fee_rate = 0.0025
quantity_to_sell = available_quantity / (1 + fee_rate)
print(f"-> 매도 수량 (수수료 고려): {quantity_to_sell:.4f} {coin_symbol}")
if (quantity_to_sell * current_price) < 5000:
print(f"-> 총 매도 금액이 5,000원 미만이라 주문을 실행하지 않습니다.")
return
# ★★★ 실제 매도 주문 실행! ★★★
bithumb_api_client.place_order(
order_currency=coin_symbol,
side="sell",
ord_type="market_sell",
units=quantity_to_sell
)
trade_info = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "type": "sell", "symbol": coin_symbol.upper(),
"quantity": quantity_to_sell, "price_per_unit": current_price
}
trade_history.add_trade_to_history(trade_info)
def _check_and_execute_real_buy_strategy(self, coin_symbol):
"""[실전 모드] 미보유 코인에 대한 매수 전략을 확인하고 실제 주문을 실행합니다."""
# 1. 기술적 분석으로 매수 신호 확인 (가상 모드와 동일)
candlestick_data = bithumb_api_client.get_candlestick_data(coin_symbol, chart_intervals=self.analysis_interval)
signal = "HOLD"
if candlestick_data and len(candlestick_data) > 20:
df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
df['close'] = pd.to_numeric(df['close'])
df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)
signal = self._check_cross_signal(df)
print(f"-> [{coin_symbol}] 미보유. 기술적 신호: {signal}")
# 2. 매수 신호 발생 시 실제 매수 주문 실행
if signal == "BUY":
print(f"-> [{coin_symbol}] 🟢 골든 크로스 발생! 실제 신규 매수를 시도합니다.")
# 실제 사용 가능한 KRW 조회
available_cash = 0.0
real_account_info = bithumb_api_client.get_account_info()
if real_account_info is None:
print("🔥 [실전 모드 에러] 계좌 정보를 가져올 수 없어 사이클을 건너뜁니다.")
return
for asset in real_account_info:
if asset.get('currency') == 'KRW':
available_cash = float(asset.get('balance'))
break
# 수수료 및 최소 주문 금액을 고려한 적응형 투자 금액 결정
fee_rate = 0.0025
cash_to_invest = 0
if available_cash <= 10000: # 1만원 이하면 전액 투자
cash_to_invest = available_cash
else:
intended_investment = available_cash * self.buy_rate
cash_to_invest = 5000 if intended_investment < 5000 else intended_investment
if cash_to_invest < 5000:
print(f"-> 보유 현금({available_cash:,.0f}원)이 최소 주문 금액(5000원)보다 적어 매수를 건너뜁니다.")
return
# 수수료를 고려한 순수 매수 금액
net_investment = cash_to_invest / (1 + fee_rate)
print(f"-> 투자 금액: {net_investment:,.0f} KRW")
# ★★★ 실제 매수 주문 실행! ★★★
bithumb_api_client.place_order(
order_currency=coin_symbol,
side="buy",
ord_type="market_buy", # 시장가 매수
price=net_investment # 시장가 매수는 총액 기준
)
quantity_to_buy = 0
current_price = 0.0
ticker = bithumb_api_client.get_ticker_info(coin_symbol)
if not ticker:
quantity_to_buy = 0
current_price = 0.0
else:
current_price = float(ticker['closing_price'])
quantity_to_buy = net_investment / current_price
trade_info = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "type": "sell", "symbol": coin_symbol.upper(),
"quantity": quantity_to_buy, "price_per_unit": current_price
}
trade_history.add_trade_to_history(trade_info)
def _check_and_settle(self):
"""중간 정산 로직"""
if self.trade_stats["total_trades"] >= self.settle_count:
odds = self.trade_stats["wins"] / self.trade_stats["total_trades"]
print("=" * 50)
print(f"⭐[중간 정산] 최근 {self.trade_stats['total_trades']}회 거래 실적⭐")
print(f"승리 횟수: {self.trade_stats['wins']} / 총 수익 금액: {self.trade_stats['total_gain']:,.0f} KRW")
print(f"패배 횟수: {self.trade_stats['losses']} / 총 손실 금액: {self.trade_stats['total_loss']:,.0f} KRW")
print(f"승률 : {odds:.2%}")
# 승률에 따른 매수 비율 조정
if odds >= 0.6 and self.buy_rate < 0.5:
old_buy_rate = self.buy_rate
self.buy_rate += 0.1
print(f" --> 높은 승률로 매수 비율이 상승🔼하였습니다. [ {old_buy_rate:.1%} --> {self.buy_rate:.1%}]")
elif odds <= 0.4 and self.buy_rate > 0.1:
old_buy_rate = self.buy_rate
self.buy_rate -= 0.1
print(f" --> 낮은 승률로 매수 비율이 하락🔽하였습니다. [ {old_buy_rate:.1%} --> {self.buy_rate:.1%}]")
print("=" * 50)
# 정산 후 통계 초기화
self.trade_stats = {"total_trades": 0, "wins": 0, "losses": 0, "total_gain": 0.0, "total_loss": 0.0}
def _print_total_assets(self):
"""총 자산 현황 출력 로직"""
print("\n" + "=" * 20 + " ⚖️ 포트폴리오 현황 ⚖️ " + "=" * 20)
total_coin_value = 0.0
for coin in self.vportfolio.get('coins_owned', []):
ticker = bithumb_api_client.get_ticker_info(coin["symbol"])
if ticker:
current_price = float(ticker['closing_price'])
coin_value = coin["quantity"] * current_price
total_coin_value += coin_value
if coin['avg_buy_price'] < 10.0 or current_price < 10.0:
print(f" - 보유: {coin['symbol']:<8} 수량: {coin['quantity']:<15.4f} 평단가: {coin['avg_buy_price']:>12,.4f} KRW | 현재가: {current_price:>12,.4f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(coin['avg_buy_price']-current_price)/coin['avg_buy_price']:>6,.1%}")
else:
print(f" - 보유: {coin['symbol']:<8} 수량: {coin['quantity']:<15.4f} 평단가: {coin['avg_buy_price']:>12,.0f} KRW | 현재가: {current_price:>12,.0f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(coin['avg_buy_price']-current_price)/coin['avg_buy_price']:>6,.1%}")
current_cash = self.vportfolio.get("cash", 0.0)
total_assets = current_cash + total_coin_value
pnl_percentage = ((total_assets - self.initial_capital) / self.initial_capital) * 100 if self.initial_capital > 0 else 0
print(f" - 현금: {current_cash:,.0f} KRW")
print("-" * 62)
print(f" - 총 자산: {total_assets:,.0f} KRW (초기 자산 대비: {pnl_percentage:+.2f}%)")
print("=" * 62)
def _print_real_total_assets(self):
real_account_info = bithumb_api_client.get_account_info()
"""[실전 모드] 실제 총 자산 현황을 출력합니다."""
print("\n" + "=" * 20 + " ⚖️ 실제 계좌 현황 ⚖️ " + "=" * 20)
if not isinstance(real_account_info, list):
print("자산 정보를 표시할 수 없습니다.")
return
total_coin_value = 0.0
current_cash = 0.0
for asset in real_account_info:
currency = asset.get('currency')
balance = float(asset.get('balance', 0))
if currency == 'KRW':
current_cash = balance
continue # KRW는 코인이 아니므로 아래 로직 건너뛰기
balance = float(asset.get('balance', 0))
locked = float(asset.get('locked', 0))
avg_buy_price = float(asset.get('avg_buy_price', 0))
available_quantity = balance - locked
if currency == "P" or available_quantity * avg_buy_price < 5000:
continue
if balance > 0:
ticker = bithumb_api_client.get_ticker_info(currency)
if ticker:
current_price = float(ticker.get('closing_price', 0))
coin_value = balance * current_price
total_coin_value += coin_value
#print(f" - 보유: {currency:<8} 수량: {balance:<15.4f} | 현재가: {current_price:>12,.0f} KRW | 평가액: {coin_value:>12,.0f} KRW")
if avg_buy_price < 10.0 or current_price < 10.0:
print(f" - 보유: {currency:<8} 수량: {balance:<15.4f} 평단가: {avg_buy_price:>12,.4f} KRW | 현재가: {current_price:>12,.4f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(current_price - avg_buy_price) / avg_buy_price:>6,.2%}")
else:
print(f" - 보유: {currency:<8} 수량: {balance:<15.4f} 평단가: {avg_buy_price:>12,.0f} KRW | 현재가: {current_price:>12,.0f} KRW | 평가액: {coin_value:>12,.0f} KRW | 수익률: {(current_price - avg_buy_price) / avg_buy_price:>6,.2%}")
total_assets = current_cash + total_coin_value
# 실전 모드의 수익률은 시작 시점의 자산을 기준으로 계산해야 하므로 initial_capital 사용
pnl_percentage = ((total_assets - self.initial_capital) / self.initial_capital) * 100 if self.initial_capital > 0 else 0
print(f" - 보유 현금: {current_cash:,.0f} KRW")
print("-" * 62)
print(f" - 총 자산: {total_assets:,.0f} KRW (초기 자산 대비: {pnl_percentage:+.2f}%)")
print("=" * 62)
def run_one_cycle(self):
"""한 번의 분석/거래 사이클 전체를 지휘하는 메인 메소드"""
# --- 쿨다운(거래 중지) 상태인지 먼저 확인! ---
is_in_cooldown = False
cooldown_until = self.vportfolio.get("cooldown_until", 0)
if time.time() < cooldown_until:
is_in_cooldown = True
resume_time_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(cooldown_until))
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ❄️ 거래 중지(쿨다운) 상태입니다. ({resume_time_str} 까지)")
elif cooldown_until > 0:
# 쿨다운 시간이 지났다면, 상태를 초기화
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ✅ 쿨다운 기간이 종료되었습니다. 거래를 재개합니다.")
self.vportfolio["consecutive_losses"] = 0
self.vportfolio["cooldown_until"] = 0
virtual_portfolio_manager.save_vportfolio(self.vportfolio) # 상태 초기화 후 저장
# 1. 분석 대상 선정
real_account_info = None
owned_symbols = []
if self.is_virtual_mode is True:
# 가상 모드: vportfolio 파일에서 보유 코인 목록 가져오기
owned_symbols = [coin['symbol'] for coin in self.vportfolio.get('coins_owned', [])]
else: # 실전 모드
real_account_info = bithumb_api_client.get_account_info()
if real_account_info is None:
print("🔥 [실전 모드 에러] 계좌 정보를 가져올 수 없어 사이클을 건너뜁니다.")
return
print("✅ [실전 모드] 실제 계좌 정보를 성공적으로 조회했습니다.")
# 실제 보유 코인 목록 생성
for asset in real_account_info:
if asset.get('currency') == 'KRW' or asset.get('currency') == 'P':
continue
balance = float(asset.get('balance', 0))
locked = float(asset.get('locked', 0))
available_quantity = balance - locked
if available_quantity * float(asset.get('avg_buy_price')) < 5000:
continue
owned_symbols.append(asset.get('currency'))
top_coins = self._filter_top_coins()
combined_list = list(set(top_coins + owned_symbols))
final_analysis_list = [coin for coin in combined_list if coin not in self.hodl_list]
print(f"\n--- 최종 분석 대상 ({len(final_analysis_list)}개) ---")
print(final_analysis_list)
print("-" * 50)
# 2. 각 코인 분석 및 거래 실행
for coin_symbol in final_analysis_list:
owned_coin_info = None
if self.is_virtual_mode:
owned_coin_info = next((c for c in self.vportfolio["coins_owned"] if c["symbol"] == coin_symbol), None)
else: # 실전 모드
owned_coin_info = next((asset for asset in real_account_info if asset.get('currency') == coin_symbol), None)
# 보유/미보유에 따른 전략 실행
if owned_coin_info:
balance = float(owned_coin_info.get('balance', 0))
locked = float(owned_coin_info.get('locked', 0))
available_quantity = balance - locked
if available_quantity * float(owned_coin_info.get('avg_buy_price')) < 5000:
continue
if self.is_virtual_mode:
self._check_and_execute_sell_strategy(coin_symbol, owned_coin_info)
else: # 실전 모드
self._check_and_execute_real_sell_strategy(coin_symbol, owned_coin_info)
elif not is_in_cooldown:
if self.is_virtual_mode:
self._check_and_execute_buy_strategy(coin_symbol)
else: # 실전 모드
self._check_and_execute_real_buy_strategy(coin_symbol)
# 3. 사이클 종료 후 중간 정산 및 자산 현황 출력
self._check_and_settle()
if self.is_virtual_mode:
self._print_total_assets()
else:
self._print_real_total_assets()
끝!!!!!!
'프로젝트 > 코인 투자 매크로' 카테고리의 다른 글
| 2025-06-21 [9] [v1.5] 연속 손실 제한 (Consecutive Loss Limit) (2) | 2025.06.21 |
|---|---|
| 2025-06-09 [9] [v1.2] 예외 종목(HODL LIST) 선택 및 수익실현/손절매 (1) | 2025.06.11 |
| 2025-06-09 [9] 종목 선택 (0) | 2025.06.10 |
| 2025-06-08 [8] [v1.0.0 완료] 캔들스틱, 단순 이동 평균(SMA), 골든/데드 크로스 (3) | 2025.06.08 |
| 2025-06-08 [7] 가상 매수/매도 함수 구현 (1) | 2025.06.08 |