feat: add shadow trading rehabilitation for WF-blocked tickers

When WF filter blocks a ticker, automatically start a virtual (shadow)
position with the same trailing/time stop logic. After WF_SHADOW_WINS
consecutive shadow wins, reset WF history to re-enable real trading.

- trader.py: add _shadow_positions, _shadow_cons_wins state;
  _shadow_enter(), get_shadow_positions(), update_shadow_peak(),
  close_shadow() functions; trigger shadow entry on WF block in buy()
- monitor.py: add _check_shadow_position() with ATR trailing + time stop;
  check shadow positions every 10s in run_monitor()
- Env: WF_SHADOW_WINS=2 (2 consecutive wins to rehabilitate)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
joungmin
2026-03-01 23:57:45 +09:00
parent 16b4c932a2
commit 29d48f0fe9
2 changed files with 246 additions and 44 deletions

View File

@@ -112,6 +112,47 @@ def _check_time_stop(ticker: str, pos: dict, current: float) -> bool:
return True
def _check_shadow_position(ticker: str, spos: dict) -> None:
"""Shadow 포지션 청산 조건 체크 (트레일링 + 타임 스탑).
실제 포지션과 동일한 로직을 적용하되 주문 없이 결과만 기록.
"""
current = get_current_price(ticker)
if current is None:
return
trader.update_shadow_peak(ticker, current)
# 갱신 후 최신 값 재조회
spos = trader.get_shadow_positions().get(ticker)
if spos is None:
return
buy_price = spos["buy_price"]
peak = spos["peak_price"]
entry_time = spos["entry_time"]
stop_pct = _get_adaptive_stop(ticker)
drop_from_peak = (peak - current) / peak
elapsed_hours = (datetime.now() - entry_time).total_seconds() / 3600
pnl_pct = (current - buy_price) / buy_price * 100
reason = None
if drop_from_peak >= stop_pct:
reason = (
f"트레일링스탑 | 최고={peak:,.2f}→현재={current:,.2f}"
f" ({drop_from_peak:.2%} | 스탑={stop_pct:.2%})"
)
elif elapsed_hours >= TIME_STOP_HOURS and pnl_pct < TIME_STOP_MIN_GAIN_PCT:
reason = (
f"타임스탑 | {elapsed_hours:.1f}h 경과 "
f"수익률={pnl_pct:+.2f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.2f}%)"
)
if reason:
trader.close_shadow(ticker, current, pnl_pct, reason)
def _check_position(ticker: str, pos: dict) -> None:
"""단일 포지션 전체 체크 (트레일링 스탑 → 타임 스탑 순서)."""
current = get_current_price(ticker)
@@ -149,10 +190,20 @@ def run_monitor(interval: int = CHECK_INTERVAL) -> None:
f"타임스탑={TIME_STOP_HOURS:.0f}h/{TIME_STOP_MIN_GAIN_PCT:+.2f}%"
)
while True:
# 실제 포지션 감시
positions_snapshot = dict(trader.get_positions())
for ticker, pos in positions_snapshot.items():
try:
_check_position(ticker, pos)
except Exception as e:
logger.error(f"모니터 오류 {ticker}: {e}")
# Shadow 포지션 감시 (WF차단 종목 재활 추적)
shadow_snapshot = trader.get_shadow_positions()
for ticker, spos in shadow_snapshot.items():
try:
_check_shadow_position(ticker, spos)
except Exception as e:
logger.error(f"Shadow 모니터 오류 {ticker}: {e}")
time.sleep(interval)