Check for positions before triggering stops

Before triggering long/short stops, check if long/short positions exist.
edtechre · Dec 11, 2023 · c802f12 · c802f12
1 parent 60416a5
commit c802f12
Show file tree

Hide file tree

Showing 2 changed files with 106 additions and 0 deletions.
diff --git a/src/pybroker/portfolio.py b/src/pybroker/portfolio.py
@@ -39,6 +39,7 @@ class Stop(NamedTuple):
     Attributes:
         id: Unique identifier.
         symbol: Symbol of the stop.
+        stop_type: :class:`.StopType`.
         pos_type: Type of  :class:`.Position`, either ``long`` or ``short``.
         percent: Percent from entry price.
         points: Cash amount from entry price.
@@ -1024,6 +1025,13 @@ def _trigger_stop(
         entry: Entry,
         stop: Stop,
     ) -> bool:
+        if stop.pos_type == "long" and stop.symbol not in self.long_positions:
+            return False
+        if (
+            stop.pos_type == "short"
+            and stop.symbol not in self.short_positions
+        ):
+            return False
         if stop.stop_type == StopType.BAR:
             fill_price = self._trigger_bar_stop(stop, price_scope, entry)
         elif (

diff --git a/tests/test_portfolio.py b/tests/test_portfolio.py
@@ -3024,6 +3024,104 @@ def test_remove_stops_when_entry(stop_type):
     assert not len(portfolio.trades)
 
 
+def test_long_stop_when_no_pos():
+    df = pd.DataFrame(
+        [
+            [SYMBOL_1, DATE_1, 75, 200, 100],
+            [SYMBOL_1, DATE_2, 250, 400, 300],
+            [SYMBOL_1, DATE_3, 290, 395, 295],
+            [SYMBOL_1, DATE_4, 200, 300, 200],
+        ],
+        columns=["symbol", "date", "open", "high", "close"],
+    )
+    df = df.set_index(["symbol", "date"])
+    sym_end_index = {SYMBOL_1: 2}
+    price_scope = PriceScope(ColumnScope(df), sym_end_index)
+    stops = (
+        Stop(
+            id=1,
+            symbol=SYMBOL_1,
+            stop_type=StopType.LOSS,
+            pos_type="long",
+            percent=10,
+            points=20,
+            bars=None,
+            fill_price=None,
+            limit_price=500,
+            exit_price=PriceType.OPEN,
+        ),
+    )
+    entry_price = Decimal(100)
+    portfolio = Portfolio(CASH)
+    portfolio.buy(
+        DATE_1,
+        SYMBOL_1,
+        SHARES_1,
+        entry_price,
+        limit_price=None,
+        stops=stops,
+    )
+    portfolio.sell(DATE_1, SYMBOL_1, SHARES_1, entry_price)
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_2, price_scope)
+    sym_end_index[SYMBOL_1] += 1
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_3, price_scope)
+    sym_end_index[SYMBOL_1] += 1
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_4, price_scope)
+    assert len(portfolio.orders) == 2
+
+
+def test_short_stop_when_no_pos():
+    df = pd.DataFrame(
+        [
+            [SYMBOL_1, DATE_1, 200, 350, 300],
+            [SYMBOL_1, DATE_2, 100, 230, 200],
+            [SYMBOL_1, DATE_3, 110, 215, 210],
+            [SYMBOL_1, DATE_4, 300, 400, 400],
+        ],
+        columns=["symbol", "date", "low", "open", "close"],
+    )
+    df = df.set_index(["symbol", "date"])
+    sym_end_index = {SYMBOL_1: 2}
+    price_scope = PriceScope(ColumnScope(df), sym_end_index)
+    stops = (
+        Stop(
+            id=1,
+            symbol=SYMBOL_1,
+            stop_type=StopType.LOSS,
+            pos_type="short",
+            percent=20,
+            points=None,
+            bars=None,
+            fill_price=None,
+            limit_price=50,
+            exit_price=PriceType.OPEN,
+        ),
+    )
+    entry_price = Decimal(300)
+    portfolio = Portfolio(CASH)
+    portfolio.sell(
+        DATE_1,
+        SYMBOL_1,
+        SHARES_1,
+        entry_price,
+        limit_price=None,
+        stops=stops,
+    )
+    portfolio.buy(DATE_1, SYMBOL_1, SHARES_1, entry_price)
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_2, price_scope)
+    sym_end_index[SYMBOL_1] += 1
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_3, price_scope)
+    sym_end_index[SYMBOL_1] += 1
+    portfolio.incr_bars()
+    portfolio.check_stops(DATE_4, price_scope)
+    assert len(portfolio.orders) == 2
+
+
 def test_win_loss_rate():
     portfolio = Portfolio(CASH)
     portfolio.buy(DATE_1, SYMBOL_1, SHARES_1, FILL_PRICE_1, LIMIT_PRICE_1)