joungmin/upbit-trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add walk-forward trade filter to prevent re-entry on losing tickers

Browse Source 
- Add trade_results table to Oracle DB for persistent trade history
- Record win/loss after each sell with pnl_pct
- Load last N trades per ticker from DB on startup (survives restarts)
- Block buy() when recent win rate (last 5 trades) < 40% threshold
- Configurable via WF_WINDOW and WF_MIN_WIN_RATE env vars
- Backtest showed improvement from -7.5% to +37.4% cumulative return

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
joungmin
2026-02-28 23:28:07 +09:00
parent 80ab004eba
commit 4888aa0faa
2 changed files with 309 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
core/trader.py
View File

@@ -12,19 +12,33 @@ from typing import Optional
import pyupbit
from dotenv import load_dotenv
from .notify import notify_buy, notify_sell, notify_error
from .price_db import (
delete_position, load_positions, upsert_position,
ensure_trade_results_table, record_trade, load_recent_wins,
)
load_dotenv()
logger = logging.getLogger(__name__)
MAX_BUDGET = 1_000_000 # 총 운용 한도: 100만원
MAX_POSITIONS = 3 # 최대 동시 보유 종목 수
PER_POSITION = MAX_BUDGET // MAX_POSITIONS # 종목당 33만3천원
MAX_BUDGET = int(os.getenv("MAX_BUDGET", "10000000")) # 총 운용 한도
MAX_POSITIONS = int(os.getenv("MAX_POSITIONS", "3")) # 최대 동시 보유 종목 수
PER_POSITION = MAX_BUDGET // MAX_POSITIONS # 종목당 투자금
# Walk-forward 필터 설정
WF_WINDOW = int(float(os.getenv("WF_WINDOW", "5"))) # 이력 윈도우 크기
WF_MIN_WIN_RATE = float(os.getenv("WF_MIN_WIN_RATE", "0.40")) # 최소 승률 임계값
_lock = threading.Lock()
_positions: dict = {}
# 구조: { ticker: { buy_price, peak_price, amount, invested_krw, entry_time } }
_last_sell_prices: dict[str, float] = {}
# 직전 매도가 기록 — 재매수 시 이 가격 이상일 때만 진입 허용
_trade_history: dict[str, list[bool]] = {}
# walk-forward 이력: { ticker: [True/False, ...] } (True=수익)
_upbit: Optional[pyupbit.Upbit] = None
@@ -35,14 +49,71 @@ def _get_upbit() -> pyupbit.Upbit:
return _upbit
def _get_history(ticker: str) -> list[bool]:
"""in-memory 이력 반환. 없으면 DB에서 초기 로드."""
if ticker not in _trade_history:
try:
_trade_history[ticker] = load_recent_wins(ticker, WF_WINDOW)
except Exception:
_trade_history[ticker] = []
return _trade_history[ticker]
def _update_history(ticker: str, is_win: bool, pnl_pct: float) -> None:
"""매도 후 in-memory 이력 갱신 + DB 기록."""
hist = _trade_history.setdefault(ticker, [])
hist.append(is_win)
# 윈도우 초과분 제거 (메모리 절약)
if len(hist) > WF_WINDOW * 2:
_trade_history[ticker] = hist[-WF_WINDOW:]
try:
record_trade(ticker, is_win, pnl_pct)
except Exception as e:
logger.error(f"거래 이력 저장 실패 {ticker}: {e}")
def _db_upsert(ticker: str, pos: dict) -> None:
"""포지션을 Oracle DB에 저장 (실패해도 거래는 계속)."""
try:
upsert_position(
ticker=ticker,
buy_price=pos["buy_price"],
peak_price=pos["peak_price"],
amount=pos["amount"],
invested_krw=pos["invested_krw"],
entry_time=pos["entry_time"].isoformat(),
)
except Exception as e:
logger.error(f"포지션 DB 저장 실패 {ticker}: {e}")
def get_positions() -> dict:
return _positions
def restore_positions() -> None:
"""시작 시 Upbit 실제 잔고를 읽어 포지션 복원 (재시작 이중 매수 방지)."""
"""시작 시 Oracle DB + Upbit 잔고를 교차 확인하여 포지션 복원.
trade_results 테이블도 이 시점에 생성 (없으면).
DB에 저장된 실제 매수가를 복원하고, Upbit 잔고에 없으면 DB에서도 삭제한다.
"""
# trade_results 테이블 초기화
try:
ensure_trade_results_table()
except Exception as e:
logger.warning(f"trade_results 테이블 생성 실패 (무시): {e}")
# DB에서 저장된 포지션 로드
try:
saved = {row["ticker"]: row for row in load_positions()}
except Exception as e:
logger.error(f"DB 포지션 로드 실패: {e}")
saved = {}
upbit = _get_upbit()
balances = upbit.get_balances()
upbit_tickers = set()
for b in balances:
currency = b["currency"]
if currency == "KRW":
@@ -57,18 +128,51 @@ def restore_positions() -> None:
invested_krw = int(amount * current)
if invested_krw < 1_000: # 소액 잔고 무시
continue
with _lock:
_positions[ticker] = {
"buy_price": current, # 정확한 매수가 불명 → 현재가로 초기화
"peak_price": current,
"amount": amount,
"invested_krw": min(invested_krw, PER_POSITION),
"entry_time": datetime.now(),
}
logger.info(
f"[복원] {ticker} 수량={amount} | 현재가={current:,.0f}원 "
f"(재시작 시 복원, 매수가 불명으로 현재가 기준)"
)
upbit_tickers.add(ticker)
if ticker in saved:
# DB에 저장된 실제 매수가 복원
s = saved[ticker]
peak = max(s["peak_price"], current) # 재시작 중 올랐을 수 있으므로 높은 쪽
entry_time = datetime.fromisoformat(s["entry_time"]) if isinstance(s["entry_time"], str) else s["entry_time"]
with _lock:
_positions[ticker] = {
"buy_price": s["buy_price"],
"peak_price": peak,
"amount": amount,
"invested_krw": s["invested_krw"],
"entry_time": entry_time,
}
logger.info(
f"[복원] {ticker} 매수가={s['buy_price']:,.0f}원 | 현재가={current:,.0f}원 "
f"| 수량={amount} (DB 복원)"
)
else:
# DB에 없음 → 현재가로 초기화 후 DB에 저장
entry_time = datetime.now()
with _lock:
_positions[ticker] = {
"buy_price": current,
"peak_price": current,
"amount": amount,
"invested_krw": min(invested_krw, PER_POSITION),
"entry_time": entry_time,
}
_db_upsert(ticker, _positions[ticker])
logger.warning(
f"[복원] {ticker} 현재가={current:,.0f}원 | 수량={amount} "
f"(DB 기록 없음 → 현재가로 초기화)"
)
# Upbit 잔고에 없는데 DB에 남아있는 항목 정리
for ticker in saved:
if ticker not in upbit_tickers:
try:
delete_position(ticker)
logger.info(f"[정리] {ticker} Upbit 잔고 없음 → DB 포지션 삭제")
except Exception as e:
logger.error(f"DB 포지션 삭제 실패 {ticker}: {e}")
def buy(ticker: str) -> bool:
@@ -78,6 +182,30 @@ def buy(ticker: str) -> bool:
logger.debug(f"{ticker} 이미 보유 중")
return False
# 직전 매도가 +1% 이상일 때만 재진입 (손절 직후 역방향 재매수 방지)
if ticker in _last_sell_prices:
current_check = pyupbit.get_current_price(ticker)
last_sell = _last_sell_prices[ticker]
threshold = last_sell * 1.01
if current_check and current_check < threshold:
logger.info(
f"[재매수 차단] {ticker} 현재={current_check:,.2f} < "
f"직전매도+1%={threshold:,.2f} → 상승 흐름 미확인"
)
return False
# Walk-forward 필터: 직전 WF_WINDOW건 승률이 낮으면 진입 차단
if WF_MIN_WIN_RATE > 0:
hist = _get_history(ticker)
if len(hist) >= WF_WINDOW:
recent_wr = sum(hist[-WF_WINDOW:]) / WF_WINDOW
if recent_wr < WF_MIN_WIN_RATE:
logger.info(
f"[WF차단] {ticker} 직전{WF_WINDOW}건 승률={recent_wr*100:.0f}%"
f" < {WF_MIN_WIN_RATE*100:.0f}% → 진입 차단"
)
return False
if len(_positions) >= MAX_POSITIONS:
logger.info(f"최대 포지션 도달({MAX_POSITIONS}), {ticker} 패스")
return False
@@ -92,7 +220,6 @@ def buy(ticker: str) -> bool:
upbit = _get_upbit()
try:
current = pyupbit.get_current_price(ticker)
result = upbit.buy_market_order(ticker, order_krw)
if not result or "error" in str(result):
logger.error(f"매수 실패: {result}")
@@ -102,18 +229,23 @@ def buy(ticker: str) -> bool:
currency = ticker.split("-")[1]
amount = float(upbit.get_balance(currency) or 0)
# 실제 체결가 = 투자금 / 수량 (시장가 주문 슬리피지 반영)
actual_price = order_krw / amount if amount > 0 else pyupbit.get_current_price(ticker)
entry_time = datetime.now()
_positions[ticker] = {
"buy_price": current,
"peak_price": current,
"buy_price": actual_price,
"peak_price": actual_price,
"amount": amount,
"invested_krw": order_krw,
"entry_time": datetime.now(),
"entry_time": entry_time,
}
_db_upsert(ticker, _positions[ticker])
logger.info(
f"[매수] {ticker} @ {current:,.0f}원 | "
f"[매수] {ticker} @ {actual_price:,.0f}원 (실체결가) | "
f"수량={amount} | 투자금={order_krw:,}원"
)
notify_buy(ticker, current, amount, order_krw)
notify_buy(ticker, actual_price, amount, order_krw)
return True
except Exception as e:
logger.error(f"매수 예외 {ticker}: {e}")
@@ -130,19 +262,41 @@ def sell(ticker: str, reason: str = "") -> bool:
pos = _positions[ticker]
upbit = _get_upbit()
try:
result = upbit.sell_market_order(ticker, pos["amount"])
currency = ticker.split("-")[1]
# 실제 잔고 확인 (재시작 후 이미 매도된 경우 대비)
actual_amount = float(upbit.get_balance(currency) or 0)
if actual_amount < 0.00001:
logger.warning(f"[매도] {ticker} 실제 잔고 없음 → 포지션 정리 (이미 매도됨)")
del _positions[ticker]
return True
result = upbit.sell_market_order(ticker, actual_amount)
if not result or "error" in str(result):
logger.error(f"매도 실패: {result}")
# 실패 후에도 잔고 재확인 → 0이면 실제로는 매도됨
actual_amount2 = float(upbit.get_balance(currency) or 0)
if actual_amount2 < 0.00001:
logger.warning(f"[매도] {ticker} 잔고 소진 확인 → 포지션 정리")
del _positions[ticker]
return True
return False
current = pyupbit.get_current_price(ticker)
pnl = (current - pos["buy_price"]) / pos["buy_price"] * 100
pnl = (current - pos["buy_price"]) / pos["buy_price"] * 100 if current else 0.0
logger.info(
f"[매도] {ticker} @ {current:,.0f}원 | "
f"수익률={pnl:+.1f}% | 사유={reason}"
)
notify_sell(ticker, current, pnl, reason)
if current:
_last_sell_prices[ticker] = current # 재매수 기준가 기록
_update_history(ticker, pnl > 0, pnl) # walk-forward 이력 갱신
del _positions[ticker]
try:
delete_position(ticker)
except Exception as e:
logger.error(f"포지션 DB 삭제 실패 {ticker}: {e}")
return True
except Exception as e:
logger.error(f"매도 예외 {ticker}: {e}")
@@ -156,3 +310,4 @@ def update_peak(ticker: str, current_price: float) -> None:
if ticker in _positions:
if current_price > _positions[ticker]["peak_price"]:
_positions[ticker]["peak_price"] = current_price
_db_upsert(ticker, _positions[ticker])