auto-sync: 2026-06-01 00:28:10

guanyu-risk/stop_loss.py

@@ -0,0 +1,49 @@
+"""
+止损模块 - 固定止损 & 追踪止损
+关羽 guanyu-dev
+"""
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class FixedStopLoss:
+    """固定止损：价格跌破 stop_price 即触发。"""
+    stop_price: float
+
+    def check(self, current_price: float) -> bool:
+        """返回 True 表示触发止损。"""
+        return current_price <= self.stop_price
+
+
+@dataclass
+class TrailingStopLoss:
+    """
+    追踪止损：
+    - 价格涨到 activation_price 以上后激活
+    - 激活后按 trail_percent 回撤触发
+    """
+    activation_price: float
+    trail_percent: float  # e.g. 0.05 = 5%
+    _high_watermark: float = field(default=0.0, init=False)
+    _activated: bool = field(default=False, init=False)
+
+    @property
+    def stop_price(self) -> float:
+        """当前止损价（未激活返回 NaN）。"""
+        if not self._activated:
+            return float("nan")
+        return self._high_watermark * (1 - self.trail_percent)
+
+    def update(self, current_price: float) -> None:
+        """更新最高价和止损线。"""
+        if current_price >= self.activation_price:
+            self._activated = True
+        if self._activated and current_price > self._high_watermark:
+            self._high_watermark = current_price
+
+    def check(self, current_price: float) -> bool:
+        """返回 True 表示触发止损。"""
+        if not self._activated:
+            return False
+        return current_price <= self.stop_price

guanyu-risk/test_stop_loss.py

@@ -0,0 +1,51 @@
+"""
+止损模块单元测试
+关羽 guanyu-dev
+"""
+
+import math
+import pytest
+from stop_loss import FixedStopLoss, TrailingStopLoss
+
+
+class TestFixedStopLoss:
+    """固定止损测试。"""
+
+    def test_triggered(self):
+        sl = FixedStopLoss(stop_price=100.0)
+        assert sl.check(99.9) is True
+
+    def test_triggered_at_stop(self):
+        sl = FixedStopLoss(stop_price=100.0)
+        assert sl.check(100.0) is True  # 等于止损价也触发
+
+    def test_not_triggered(self):
+        sl = FixedStopLoss(stop_price=100.0)
+        assert sl.check(100.1) is False
+
+
+class TestTrailingStopLoss:
+    """追踪止损测试。"""
+
+    def test_not_activated_below_activation(self):
+        tsl = TrailingStopLoss(activation_price=100.0, trail_percent=0.05)
+        tsl.update(95.0)
+        assert tsl.check(95.0) is False
+
+    def test_activated_and_update_high(self):
+        tsl = TrailingStopLoss(activation_price=100.0, trail_percent=0.05)
+        tsl.update(100.0)  # activate, high=100
+        tsl.update(110.0)  # high=110, stop=104.5
+        assert math.isclose(tsl.stop_price, 104.5)
+
+    def test_triggered_after_pullback(self):
+        tsl = TrailingStopLoss(activation_price=100.0, trail_percent=0.05)
+        tsl.update(100.0)  # activate, high=100
+        tsl.update(110.0)  # high=110, stop=104.5
+        assert tsl.check(104.4) is True
+
+    def test_not_triggered_in_trail(self):
+        tsl = TrailingStopLoss(activation_price=100.0, trail_percent=0.05)
+        tsl.update(100.0)
+        tsl.update(110.0)  # stop=104.5
+        assert tsl.check(105.0) is False