왜 실시간 PnL 추적이 필요한가
자동매매 봇을 운영하면서 “지금 얼마 벌고 있는지”를 정확히 모르는 경우가 의외로 많습니다. 거래소 잔고만 확인하면 미실현 손익(Unrealized PnL)이 빠지고, 체결 내역만 보면 수수료와 펀딩비가 반영되지 않습니다. 이 글에서는 자동매매 봇에 실시간 PnL 추적 시스템을 구현하는 방법을 단계별로 다룹니다.
PnL의 세 가지 구성 요소
정확한 손익 추적을 위해 반드시 구분해야 할 세 가지 요소가 있습니다.
- 실현 손익(Realized PnL): 이미 청산한 포지션에서 확정된 수익/손실. 매수가와 매도가의 차이에 수량을 곱한 값입니다.
- 미실현 손익(Unrealized PnL): 아직 보유 중인 포지션의 평가 손익. 현재 시장가와 진입가의 차이로 계산합니다.
- 비용(Costs): 거래 수수료, 선물 펀딩비, 슬리피지 등 매매에 수반되는 모든 비용입니다.
순 PnL = 실현 손익 + 미실현 손익 – 총 비용. 이 공식을 정확히 구현하는 것이 핵심입니다.
포지션 트래커 설계
PnL을 정확히 추적하려면 각 포지션의 진입가, 수량, 방향을 기록하는 포지션 트래커가 필요합니다.
from dataclasses import dataclass, field
from typing import Dict, List
import time
@dataclass
class Position:
symbol: str
side: str # "LONG" or "SHORT"
qty: float = 0.0
avg_entry: float = 0.0
realized_pnl: float = 0.0
total_fees: float = 0.0
trades: List[dict] = field(default_factory=list)
def add_fill(self, price: float, qty: float, fee: float):
"""체결 추가 및 평균 진입가 갱신"""
if self.qty == 0:
self.avg_entry = price
self.qty = qty
else:
# 같은 방향 추가 진입
total_cost = self.avg_entry * self.qty + price * qty
self.qty += qty
self.avg_entry = total_cost / self.qty if self.qty > 0 else 0
self.total_fees += fee
self.trades.append({
"price": price, "qty": qty, "fee": fee,
"timestamp": time.time()
})
def reduce(self, price: float, qty: float, fee: float):
"""포지션 축소 및 실현 손익 계산"""
if self.side == "LONG":
pnl = (price - self.avg_entry) * qty
else:
pnl = (self.avg_entry - price) * qty
self.realized_pnl += pnl
self.total_fees += fee
self.qty -= qty
if self.qty <= 0.0001:
self.qty = 0.0
self.avg_entry = 0.0
return pnl
def unrealized_pnl(self, current_price: float) -> float:
"""미실현 손익 계산"""
if self.qty == 0:
return 0.0
if self.side == "LONG":
return (current_price - self.avg_entry) * self.qty
else:
return (self.avg_entry - current_price) * self.qty
add_fill은 동일 방향 추가 진입 시 가중평균 진입가를 계산합니다. reduce는 포지션 축소 시 실현 손익을 확정합니다. 이 두 메서드가 PnL 추적의 핵심입니다.
실시간 PnL 대시보드 클래스
여러 심볼의 포지션을 통합 관리하고 전체 PnL을 실시간으로 계산하는 대시보드를 만듭니다.
class PnLDashboard:
def __init__(self, initial_balance: float):
self.initial_balance = initial_balance
self.positions: Dict[str, Position] = {}
self.closed_pnl = 0.0
self.total_fees = 0.0
self.funding_fees = 0.0
self.snapshots: List[dict] = []
def get_or_create(self, symbol: str, side: str) -> Position:
if symbol not in self.positions:
self.positions[symbol] = Position(symbol=symbol, side=side)
return self.positions[symbol]
def record_funding(self, symbol: str, amount: float):
"""펀딩비 기록 (선물 거래)"""
self.funding_fees += amount
def total_unrealized(self, prices: Dict[str, float]) -> float:
"""전체 미실현 손익"""
return sum(
pos.unrealized_pnl(prices.get(pos.symbol, 0))
for pos in self.positions.values()
if pos.qty > 0
)
def net_pnl(self, prices: Dict[str, float]) -> float:
"""순 PnL = 실현 + 미실현 - 수수료 - 펀딩비"""
realized = sum(p.realized_pnl for p in self.positions.values())
unrealized = self.total_unrealized(prices)
fees = sum(p.total_fees for p in self.positions.values())
return realized + unrealized - fees - self.funding_fees
def roi_percent(self, prices: Dict[str, float]) -> float:
"""수익률(%) 계산"""
if self.initial_balance == 0:
return 0.0
return (self.net_pnl(prices) / self.initial_balance) * 100
def take_snapshot(self, prices: Dict[str, float]):
"""현재 상태 스냅샷 저장 (차트용)"""
self.snapshots.append({
"timestamp": time.time(),
"net_pnl": self.net_pnl(prices),
"roi": self.roi_percent(prices),
"positions": len([p for p in self.positions.values() if p.qty > 0])
})
take_snapshot은 주기적으로 호출하여 PnL 곡선을 그릴 수 있는 시계열 데이터를 생성합니다. 1분 간격으로 저장하면 일간 PnL 차트를 만들기에 충분합니다.
MDD(최대 낙폭) 실시간 추적
PnL과 함께 반드시 추적해야 할 지표가 MDD(Maximum Drawdown)입니다. 봇이 운영 중 최고점 대비 얼마나 하락했는지를 실시간으로 알아야 리스크를 통제할 수 있습니다. MDD 제어에 대한 심화 내용은 자동매매 MDD 제어 전략을 참고하세요.
class DrawdownTracker:
def __init__(self):
self.peak_equity = 0.0
self.max_drawdown = 0.0
self.current_drawdown = 0.0
def update(self, current_equity: float):
if current_equity > self.peak_equity:
self.peak_equity = current_equity
if self.peak_equity > 0:
self.current_drawdown = (self.peak_equity - current_equity) / self.peak_equity
self.max_drawdown = max(self.max_drawdown, self.current_drawdown)
def should_stop(self, threshold: float = 0.10) -> bool:
"""MDD가 임계값 초과 시 매매 중단 신호"""
return self.current_drawdown >= threshold
threshold=0.10은 10% MDD에서 매매를 중단하는 설정입니다. 전략과 리스크 허용도에 따라 5~20% 범위에서 조절합니다.
수수료 정확히 반영하기
PnL 추적에서 가장 흔한 실수는 수수료를 과소 계산하는 것입니다. 거래소별로 수수료 체계가 다르고, 메이커/테이커 수수료도 다릅니다.
FEE_RATES = {
"binance_spot": {"maker": 0.001, "taker": 0.001},
"binance_futures": {"maker": 0.0002, "taker": 0.0004},
"bybit_futures": {"maker": 0.0001, "taker": 0.0006},
}
def calculate_fee(exchange: str, order_type: str, notional: float) -> float:
rates = FEE_RATES.get(exchange, {"maker": 0.001, "taker": 0.001})
rate = rates["maker"] if order_type == "limit" else rates["taker"]
return notional * rate
하루 100회 거래하는 스캘핑 봇이라면 수수료만으로 일일 원금의 4~6%가 빠질 수 있습니다. 이를 PnL에 정확히 반영하지 않으면 수익이 나는 것처럼 보이지만 실제로는 손실인 상황이 발생합니다.
텔레그램 실시간 리포트
PnL 데이터를 수집했다면 텔레그램으로 주기적인 리포트를 보낼 수 있습니다. 알림 구현에 대한 기본 설정은 자동매매 텔레그램 알림 구현에서 다루고 있습니다.
def format_pnl_report(dashboard: PnLDashboard, dd: DrawdownTracker, prices: dict) -> str:
net = dashboard.net_pnl(prices)
roi = dashboard.roi_percent(prices)
active = len([p for p in dashboard.positions.values() if p.qty > 0])
emoji = "🟢" if net >= 0 else "🔴"
return (
f"{emoji} PnL 리포트n"
f"━━━━━━━━━━━━━━━n"
f"순 PnL: ${net:,.2f} ({roi:+.2f}%)n"
f"MDD: {dd.max_drawdown:.2%}n"
f"현재 DD: {dd.current_drawdown:.2%}n"
f"활성 포지션: {active}개n"
f"총 수수료: ${dashboard.total_fees:.2f}n"
f"펀딩비: ${dashboard.funding_fees:.2f}"
)
마무리
실시간 PnL 추적은 자동매매의 계기판입니다. 계기판 없이 운전하는 것이 위험하듯, PnL을 모르고 봇을 운영하는 것은 눈을 감고 매매하는 것과 같습니다. 실현/미실현 손익 분리, 수수료·펀딩비 정확 반영, MDD 실시간 추적 — 이 세 가지만 제대로 구현해도 봇 운영의 투명성이 크게 향상됩니다. 특히 수수료 과소 계산은 많은 트레이더가 간과하는 함정이니 반드시 체크하시기 바랍니다.