diff --git a/src/config.py b/src/config.py index 656044c..7f0a367 100644 --- a/src/config.py +++ b/src/config.py @@ -62,6 +62,9 @@ class Settings(BaseSettings): USD_BUFFER_MIN: float = Field(default=1000.0, ge=0.0) US_MIN_PRICE: float = Field(default=5.0, ge=0.0) STOPLOSS_REENTRY_COOLDOWN_MINUTES: int = Field(default=120, ge=1, le=1440) + KR_ATR_STOP_MULTIPLIER_K: float = Field(default=2.0, ge=0.1, le=10.0) + KR_ATR_STOP_MIN_PCT: float = Field(default=-2.0, le=0.0) + KR_ATR_STOP_MAX_PCT: float = Field(default=-7.0, le=0.0) OVERNIGHT_EXCEPTION_ENABLED: bool = True # Trading frequency mode (daily = batch API calls, realtime = per-stock calls) diff --git a/src/main.py b/src/main.py index 2499550..bc9a926 100644 --- a/src/main.py +++ b/src/main.py @@ -119,6 +119,27 @@ def _resolve_sell_qty_for_pnl(*, sell_qty: int | None, buy_qty: int | None) -> i return max(0, int(buy_qty or 0)) +def _compute_kr_dynamic_stop_loss_pct( + *, + entry_price: float, + atr_value: float, + fallback_stop_loss_pct: float, + settings: Settings | None, +) -> float: + """Compute KR dynamic hard-stop threshold in percent.""" + if entry_price <= 0 or atr_value <= 0: + return fallback_stop_loss_pct + + k = float(getattr(settings, "KR_ATR_STOP_MULTIPLIER_K", 2.0) if settings else 2.0) + min_pct = float(getattr(settings, "KR_ATR_STOP_MIN_PCT", -2.0) if settings else -2.0) + max_pct = float(getattr(settings, "KR_ATR_STOP_MAX_PCT", -7.0) if settings else -7.0) + if max_pct > min_pct: + min_pct, max_pct = max_pct, min_pct + + dynamic_stop_pct = -((k * atr_value) / entry_price) * 100.0 + return max(max_pct, min(min_pct, dynamic_stop_pct)) + + def _stoploss_cooldown_key(*, market: MarketInfo, stock_code: str) -> str: return f"{market.code}:{stock_code}" @@ -518,6 +539,7 @@ def _apply_staged_exit_override_for_hold( open_position: dict[str, Any] | None, market_data: dict[str, Any], stock_playbook: Any | None, + settings: Settings | None = None, ) -> TradeDecision: """Apply v2 staged exit semantics for HOLD positions using runtime state.""" if decision.action != "HOLD" or not open_position: @@ -533,6 +555,14 @@ def _apply_staged_exit_override_for_hold( if stock_playbook and stock_playbook.scenarios: stop_loss_threshold = stock_playbook.scenarios[0].stop_loss_pct take_profit_threshold = stock_playbook.scenarios[0].take_profit_pct + atr_value = safe_float(market_data.get("atr_value"), 0.0) + if market.code == "KR": + stop_loss_threshold = _compute_kr_dynamic_stop_loss_pct( + entry_price=entry_price, + atr_value=atr_value, + fallback_stop_loss_pct=stop_loss_threshold, + settings=settings, + ) runtime_key = _build_runtime_position_key( market_code=market.code, @@ -558,7 +588,7 @@ def _apply_staged_exit_override_for_hold( current_price=current_price, entry_price=entry_price, peak_price=peak_price, - atr_value=safe_float(market_data.get("atr_value"), 0.0), + atr_value=atr_value, pred_down_prob=safe_float(market_data.get("pred_down_prob"), 0.0), liquidity_weak=safe_float(market_data.get("volume_ratio"), 1.0) < 1.0, ), @@ -1375,6 +1405,7 @@ async def trading_cycle( open_position=open_position, market_data=market_data, stock_playbook=stock_playbook, + settings=settings, ) if open_position and decision.action == "HOLD" and _should_force_exit_for_overnight( market=market, @@ -2582,6 +2613,7 @@ async def run_daily_session( open_position=daily_open, market_data=stock_data, stock_playbook=stock_playbook, + settings=settings, ) if daily_open and decision.action == "HOLD" and _should_force_exit_for_overnight( market=market, diff --git a/tests/test_main.py b/tests/test_main.py index 4de28aa..e98a659 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -32,6 +32,7 @@ from src.main import ( _run_context_scheduler, _run_evolution_loop, _start_dashboard_server, + _compute_kr_dynamic_stop_loss_pct, handle_domestic_pending_orders, handle_overseas_pending_orders, process_blackout_recovery_orders, @@ -135,6 +136,51 @@ def test_resolve_sell_qty_for_pnl_uses_buy_qty_fallback_when_sell_qty_missing() def test_resolve_sell_qty_for_pnl_returns_zero_when_both_missing() -> None: assert _resolve_sell_qty_for_pnl(sell_qty=None, buy_qty=None) == 0 + +def test_compute_kr_dynamic_stop_loss_pct_falls_back_without_atr() -> None: + out = _compute_kr_dynamic_stop_loss_pct( + entry_price=100.0, + atr_value=0.0, + fallback_stop_loss_pct=-2.0, + settings=None, + ) + assert out == -2.0 + + +def test_compute_kr_dynamic_stop_loss_pct_clamps_to_min_and_max() -> None: + # Small ATR -> clamp to min (-2%) + out_small = _compute_kr_dynamic_stop_loss_pct( + entry_price=100.0, + atr_value=0.2, + fallback_stop_loss_pct=-2.0, + settings=None, + ) + assert out_small == -2.0 + + # Large ATR -> clamp to max (-7%) + out_large = _compute_kr_dynamic_stop_loss_pct( + entry_price=100.0, + atr_value=10.0, + fallback_stop_loss_pct=-2.0, + settings=None, + ) + assert out_large == -7.0 + + +def test_compute_kr_dynamic_stop_loss_pct_uses_settings_values() -> None: + settings = MagicMock( + KR_ATR_STOP_MULTIPLIER_K=3.0, + KR_ATR_STOP_MIN_PCT=-1.5, + KR_ATR_STOP_MAX_PCT=-6.0, + ) + out = _compute_kr_dynamic_stop_loss_pct( + entry_price=100.0, + atr_value=1.0, + fallback_stop_loss_pct=-2.0, + settings=settings, + ) + assert out == -3.0 + def test_returns_zero_when_field_empty_string(self) -> None: """Returns 0.0 when pchs_avg_pric is an empty string.""" balance = {"output1": [{"pdno": "005930", "pchs_avg_pric": ""}]}