hyowons
216 lines
9.4 KiB
Python
216 lines
9.4 KiB
Python
"""가상 포트폴리오 — 현금·포지션·체결·비용·손익·MDD.
|
|
|
|
상태: state/sim/portfolio.json
|
|
거래로그: state/sim/trades.jsonl (append)
|
|
실제 주문 X — 전부 장부상 가상 체결.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from . import config
|
|
from .signals import target_value as _sizing_value # buy()의 signals 파라미터가 모듈명을 가려서 직접 import
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(config.KST).isoformat()
|
|
|
|
|
|
class Portfolio:
|
|
def __init__(self, data: dict):
|
|
self.cash: float = data.get('cash', config.INITIAL_CAPITAL)
|
|
self.initial: float = data.get('initial', config.INITIAL_CAPITAL)
|
|
self.positions: dict[str, dict] = data.get('positions', {})
|
|
self.realized_pnl: float = data.get('realized_pnl', 0.0)
|
|
self.closed_trades: int = data.get('closed_trades', 0)
|
|
self.wins: int = data.get('wins', 0)
|
|
self.peak_equity: float = data.get('peak_equity', self.initial)
|
|
self.max_drawdown: float = data.get('max_drawdown', 0.0)
|
|
self.created_at: str = data.get('created_at', _now_iso())
|
|
self.updated_at: str = data.get('updated_at', self.created_at)
|
|
self.last_exit: dict[str, str] = data.get('last_exit', {}) # code → 전량청산 날짜(YYYY-MM-DD), 당일 재진입 금지용
|
|
# 영속화 경로 (변이는 별도 경로, 기본은 메인 sim)
|
|
self._pf_path = config.PORTFOLIO_PATH
|
|
self._trades_path = config.TRADES_PATH
|
|
|
|
# ---- 영속화 ----
|
|
@classmethod
|
|
def load(cls, portfolio_path=None, trades_path=None) -> 'Portfolio':
|
|
pp = portfolio_path or config.PORTFOLIO_PATH
|
|
tp = trades_path or config.TRADES_PATH
|
|
data = {}
|
|
if pp.exists():
|
|
try:
|
|
data = json.loads(pp.read_text())
|
|
except Exception:
|
|
data = {}
|
|
obj = cls(data)
|
|
obj._pf_path = pp
|
|
obj._trades_path = tp
|
|
return obj
|
|
|
|
def save(self):
|
|
self._pf_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.updated_at = _now_iso()
|
|
self._pf_path.write_text(json.dumps({
|
|
'cash': self.cash, 'initial': self.initial, 'positions': self.positions,
|
|
'realized_pnl': self.realized_pnl, 'closed_trades': self.closed_trades,
|
|
'wins': self.wins, 'peak_equity': self.peak_equity,
|
|
'max_drawdown': self.max_drawdown, 'last_exit': self.last_exit,
|
|
'created_at': self.created_at, 'updated_at': self.updated_at,
|
|
}, ensure_ascii=False, indent=2))
|
|
|
|
def _log_trade(self, rec: dict):
|
|
self._trades_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self._trades_path.open('a') as f:
|
|
f.write(json.dumps(rec, ensure_ascii=False) + '\n')
|
|
|
|
# ---- 체결 ----
|
|
def can_open(self) -> bool:
|
|
return len(self.positions) < config.MAX_POSITIONS
|
|
|
|
def target_position_value(self) -> float:
|
|
return self.equity_estimate() / config.MAX_POSITIONS
|
|
|
|
def equity_estimate(self) -> float:
|
|
"""현재 포지션의 진입가 기준 추정 자산 (사이징용 — mark 전 호출 대비)."""
|
|
held = sum(p['qty'] * p.get('cur_price', p['entry_price']) for p in self.positions.values())
|
|
return self.cash + held
|
|
|
|
def _qty_for(self, fill_price: float, value: float) -> int:
|
|
return int(value // (fill_price * (1 + config.COMMISSION_RATE)))
|
|
|
|
def buy(self, code, name, fill_price, sources, stop, target, path, reason, signals=None) -> dict | None:
|
|
"""신규 진입(1차 트랜치). 목표비중을 ENTRY_TRANCHES 로 나눈 만큼만 매수. 체결 dict 또는 None."""
|
|
if code in self.positions or not self.can_open():
|
|
return None
|
|
full_target = _sizing_value(self.equity_estimate(), fill_price, stop)
|
|
tranche_val = full_target / max(1, config.ENTRY_TRANCHES)
|
|
qty = self._qty_for(fill_price, tranche_val)
|
|
if qty < 1:
|
|
return None
|
|
cost = qty * fill_price * (1 + config.COMMISSION_RATE)
|
|
if cost > self.cash:
|
|
qty = self._qty_for(fill_price, self.cash)
|
|
if qty < 1:
|
|
return None
|
|
cost = qty * fill_price * (1 + config.COMMISSION_RATE)
|
|
self.cash -= cost
|
|
self.positions[code] = {
|
|
'code': code, 'name': name, 'sources': sources,
|
|
'qty': qty, 'entry_price': fill_price, 'entry_at': _now_iso(),
|
|
'stop': stop, 'target': target, 'peak': fill_price,
|
|
'trailing_on': False, 'scaled_out': False, 'entry_path': path,
|
|
'entry_reason': reason, 'cur_price': fill_price,
|
|
'tranches': 1, 'tranche_value': tranche_val, 'last_add_price': fill_price,
|
|
}
|
|
rec = {'ts': _now_iso(), 'side': 'BUY', 'code': code, 'name': name,
|
|
'price': fill_price, 'qty': qty, 'cost': round(cost),
|
|
'path': path, 'reason': reason, 'stop': stop, 'target': target,
|
|
'signals': signals or []}
|
|
self._log_trade(rec)
|
|
return rec
|
|
|
|
def add_tranche(self, code, fill_price, reason, signals=None) -> dict | None:
|
|
"""추격매수 — 보유 종목에 다음 트랜치 추가. 평단·수량 갱신. 체결 dict 또는 None."""
|
|
pos = self.positions.get(code)
|
|
if not pos or pos.get('tranches', 1) >= config.ENTRY_TRANCHES:
|
|
return None
|
|
tranche_val = pos.get('tranche_value') or (pos['entry_price'] * pos['qty'])
|
|
qty = self._qty_for(fill_price, tranche_val)
|
|
if qty < 1:
|
|
return None
|
|
cost = qty * fill_price * (1 + config.COMMISSION_RATE)
|
|
if cost > self.cash:
|
|
qty = self._qty_for(fill_price, self.cash)
|
|
if qty < 1:
|
|
return None
|
|
cost = qty * fill_price * (1 + config.COMMISSION_RATE)
|
|
new_qty = pos['qty'] + qty
|
|
pos['entry_price'] = (pos['entry_price'] * pos['qty'] + fill_price * qty) / new_qty
|
|
pos['qty'] = new_qty
|
|
pos['tranches'] = pos.get('tranches', 1) + 1
|
|
pos['last_add_price'] = fill_price
|
|
pos['cur_price'] = fill_price
|
|
self.cash -= cost
|
|
rec = {'ts': _now_iso(), 'side': 'BUY', 'code': code, 'name': pos['name'],
|
|
'price': fill_price, 'qty': qty, 'cost': round(cost),
|
|
'path': 'add', 'reason': reason, 'stop': pos['stop'], 'target': pos['target'],
|
|
'signals': signals or []}
|
|
self._log_trade(rec)
|
|
return rec
|
|
|
|
def sell(self, code, fill_price, reason, signals=None, frac: float = 1.0) -> dict | None:
|
|
"""매도. frac<1 이면 부분(분할)매도 — 포지션 유지하고 수량만 차감."""
|
|
pos = self.positions.get(code)
|
|
if not pos:
|
|
return None
|
|
total = pos['qty']
|
|
qty = total if frac >= 1.0 else max(1, int(total * frac))
|
|
if qty >= total:
|
|
qty = total
|
|
entry = pos['entry_price']
|
|
proceeds = qty * fill_price * (1 - config.COMMISSION_RATE - config.SELL_TAX_RATE)
|
|
entry_cost = qty * entry * (1 + config.COMMISSION_RATE)
|
|
pnl = proceeds - entry_cost
|
|
self.cash += proceeds
|
|
self.realized_pnl += pnl
|
|
partial = qty < total
|
|
rec = {'ts': _now_iso(), 'side': 'SELL', 'code': code, 'name': pos['name'],
|
|
'price': fill_price, 'qty': qty, 'proceeds': round(proceeds),
|
|
'entry_price': round(entry), 'pnl': round(pnl),
|
|
'pnl_pct': round((fill_price / entry - 1) * 100, 2),
|
|
'hold_from': pos['entry_at'], 'reason': reason, 'partial': partial,
|
|
'signals': signals or []}
|
|
self._log_trade(rec)
|
|
if partial:
|
|
pos['qty'] = total - qty
|
|
else:
|
|
self.closed_trades += 1
|
|
if pnl > 0:
|
|
self.wins += 1
|
|
del self.positions[code]
|
|
self.last_exit[code] = _now_iso()[:10] # 당일 재진입 금지 기준
|
|
return rec
|
|
|
|
def exited_today(self, code: str) -> bool:
|
|
"""오늘 전량청산한 종목인지 — 당일 재진입(휩쏘 churn) 방지."""
|
|
return self.last_exit.get(code) == _now_iso()[:10]
|
|
|
|
def apply_position_update(self, code: str, upd: dict):
|
|
if code in self.positions:
|
|
self.positions[code].update(upd)
|
|
|
|
def mark(self, code: str, cur_price: int):
|
|
if code in self.positions:
|
|
self.positions[code]['cur_price'] = cur_price
|
|
|
|
# ---- 지표 ----
|
|
def update_equity_metrics(self):
|
|
eq = self.equity_estimate()
|
|
if eq > self.peak_equity:
|
|
self.peak_equity = eq
|
|
dd = (self.peak_equity - eq) / self.peak_equity if self.peak_equity else 0.0
|
|
if dd > self.max_drawdown:
|
|
self.max_drawdown = dd
|
|
return eq
|
|
|
|
def summary(self) -> dict:
|
|
eq = self.equity_estimate()
|
|
held_val = eq - self.cash
|
|
return {
|
|
'initial': self.initial,
|
|
'cash': round(self.cash),
|
|
'held_value': round(held_val),
|
|
'equity': round(eq),
|
|
'total_return_pct': round((eq / self.initial - 1) * 100, 2),
|
|
'realized_pnl': round(self.realized_pnl),
|
|
'closed_trades': self.closed_trades,
|
|
'wins': self.wins,
|
|
'win_rate_pct': round(self.wins / self.closed_trades * 100, 1) if self.closed_trades else None,
|
|
'open_positions': len(self.positions),
|
|
'max_drawdown_pct': round(self.max_drawdown * 100, 2),
|
|
'updated_at': self.updated_at,
|
|
}
|