83bd51117f
Strategy C: volatility breakout (Larry Williams K=0.5) AND momentum (MA20 + 2x volume surge) must both trigger for a buy signal. Hard rules: - Trailing stop: sell when price drops -10% from peak - Max budget: 1,000,000 KRW total, up to 3 positions (333,333 KRW each) - Scan top 20 KRW tickers by 24h trading volume every 60s - Monitor positions every 10s Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
"""매수/매도 실행 및 포지션 관리."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from typing import Optional
|
|
|
|
import pyupbit
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_BUDGET = 1_000_000 # 총 운용 한도: 100만원
|
|
MAX_POSITIONS = 3 # 최대 동시 보유 종목 수
|
|
PER_POSITION = MAX_BUDGET // MAX_POSITIONS # 종목당 33만3천원
|
|
|
|
_lock = threading.Lock()
|
|
_positions: dict = {}
|
|
# 구조: { ticker: { buy_price, peak_price, amount, invested_krw } }
|
|
|
|
_upbit: Optional[pyupbit.Upbit] = None
|
|
|
|
|
|
def _get_upbit() -> pyupbit.Upbit:
|
|
global _upbit
|
|
if _upbit is None:
|
|
_upbit = pyupbit.Upbit(os.getenv("ACCESS_KEY"), os.getenv("SECRET_KEY"))
|
|
return _upbit
|
|
|
|
|
|
def get_positions() -> dict:
|
|
return _positions
|
|
|
|
|
|
def buy(ticker: str) -> bool:
|
|
"""시장가 매수. 예산·포지션 수 확인 후 진입."""
|
|
with _lock:
|
|
if ticker in _positions:
|
|
logger.debug(f"{ticker} 이미 보유 중")
|
|
return False
|
|
|
|
if len(_positions) >= MAX_POSITIONS:
|
|
logger.info(f"최대 포지션 도달({MAX_POSITIONS}), {ticker} 패스")
|
|
return False
|
|
|
|
invested = sum(p["invested_krw"] for p in _positions.values())
|
|
available = MAX_BUDGET - invested
|
|
order_krw = min(available, PER_POSITION)
|
|
|
|
if order_krw < 10_000:
|
|
logger.info(f"잔여 예산 부족({order_krw:,}원), {ticker} 패스")
|
|
return False
|
|
|
|
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}")
|
|
return False
|
|
|
|
time.sleep(0.5) # 체결 대기
|
|
currency = ticker.split("-")[1]
|
|
amount = float(upbit.get_balance(currency) or 0)
|
|
|
|
_positions[ticker] = {
|
|
"buy_price": current,
|
|
"peak_price": current,
|
|
"amount": amount,
|
|
"invested_krw": order_krw,
|
|
}
|
|
logger.info(
|
|
f"[매수] {ticker} @ {current:,.0f}원 | "
|
|
f"수량={amount} | 투자금={order_krw:,}원"
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"매수 예외 {ticker}: {e}")
|
|
return False
|
|
|
|
|
|
def sell(ticker: str, reason: str = "") -> bool:
|
|
"""시장가 전량 매도."""
|
|
with _lock:
|
|
if ticker not in _positions:
|
|
return False
|
|
|
|
pos = _positions[ticker]
|
|
upbit = _get_upbit()
|
|
try:
|
|
result = upbit.sell_market_order(ticker, pos["amount"])
|
|
if not result or "error" in str(result):
|
|
logger.error(f"매도 실패: {result}")
|
|
return False
|
|
|
|
current = pyupbit.get_current_price(ticker)
|
|
pnl = (current - pos["buy_price"]) / pos["buy_price"] * 100
|
|
logger.info(
|
|
f"[매도] {ticker} @ {current:,.0f}원 | "
|
|
f"수익률={pnl:+.1f}% | 사유={reason}"
|
|
)
|
|
del _positions[ticker]
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"매도 예외 {ticker}: {e}")
|
|
return False
|
|
|
|
|
|
def update_peak(ticker: str, current_price: float) -> None:
|
|
"""최고가 갱신 (트레일링 스탑 기준선 상향)."""
|
|
with _lock:
|
|
if ticker in _positions:
|
|
if current_price > _positions[ticker]["peak_price"]:
|
|
_positions[ticker]["peak_price"] = current_price
|