Add fixed-direction switch for perp/option open side.

When enabled, lock perp long→buy Put or short→buy Call with ITM/ATM only; off keeps ATM/ask rules.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/strategy/session.py

@@ -16,6 +16,7 @@ from .selection import (
     atm_allows_open,
     atm_open_offset,
     hours_until_expiry,
+    is_itm_or_atm,
     list_eligible_expiry_ymds,
     option_leverage,
     select_option_pair,
@@ -101,6 +102,36 @@ def _strategy_floats() -> tuple[float, float, float, bool]:
         )
 
 
+def _fixed_direction() -> tuple[bool, str]:
+    """(enabled, perp_side long|short)。默认关。"""
+    s = get_settings()
+    try:
+        from ..models.db import get_db
+
+        db = get_db()
+        enabled = _as_bool_setting(
+            db.get_setting(
+                "fixed_direction_enabled", str(s.fixed_direction_enabled)
+            ),
+            s.fixed_direction_enabled,
+        )
+        side = str(
+            db.get_setting("fixed_perp_side", s.fixed_perp_side) or s.fixed_perp_side
+        ).strip().lower()
+        if side not in ("long", "short"):
+            side = "long"
+        return enabled, side
+    except Exception:
+        side = str(s.fixed_perp_side or "long").strip().lower()
+        if side not in ("long", "short"):
+            side = "long"
+        return bool(s.fixed_direction_enabled), side
+
+
+def _option_side_for_perp(perp_side: str) -> str:
+    return "put" if (perp_side or "").strip().lower() == "long" else "call"
+
+
 @dataclass(slots=True)
 class OpenPick:
     pair: OptionPair
@@ -230,19 +261,29 @@ class StrategySession:
         if mark is None or mark <= 0:
             raise RuntimeError("无法获取标的标记/指数价格，无法选 ATM")
         min_hours, _, _, _ = _strategy_floats()
+        fixed_on, fixed_perp = _fixed_direction()
+        opt_side = _option_side_for_perp(fixed_perp) if fixed_on else None
         contracts = self.ex.list_option_contracts(s.option_inst_family)
-        pair = select_option_pair(contracts, mark_px=float(mark), min_hours=min_hours)
+        pair = select_option_pair(
+            contracts,
+            mark_px=float(mark),
+            min_hours=min_hours,
+            option_side=opt_side,
+        )
         if pair is None:
+            kind = f"实值/平值 {opt_side}" if opt_side else "ATM"
             raise RuntimeError(
-                f"未找到剩余≥{min_hours}h 的 ATM Call/Put (family={s.option_inst_family})"
+                f"未找到剩余≥{min_hours}h 的 {kind} Call/Put (family={s.option_inst_family})"
             )
         return self._apply_pair(pair, mark=float(mark), idx=idx)
 
     def pick_for_open(self) -> OpenPick | None:
-        from .signal import decide
+        from .signal import decide, decide_fixed
 
         s = self.settings
         min_hours, min_lev, max_atm_off, atm_off_on = _strategy_floats()
+        fixed_on, fixed_perp = _fixed_direction()
+        opt_side_hint = _option_side_for_perp(fixed_perp) if fixed_on else None
         idx = self.ex.fetch_index(s.index_inst_id)
         mark = self.ex.fetch_mark(s.perp_inst_id) or idx
         if mark is None or mark <= 0:
@@ -255,24 +296,44 @@ class StrategySession:
             return None
 
         for ymd in eligible:
-            pair = select_option_pair(contracts, mark_px=underlying, expiry_ymd=ymd)
+            pair = select_option_pair(
+                contracts,
+                mark_px=underlying,
+                expiry_ymd=ymd,
+                option_side=opt_side_hint,
+            )
             if pair is None:
                 continue
-            offset = atm_open_offset(pair.strike, underlying)
-            if not atm_allows_open(
-                pair.strike,
-                underlying,
-                max_offset=max_atm_off,
-                enabled=atm_off_on,
-            ):
-                logger.info(
-                    "skip expiry=%s strike=%.0f atm_offset=%.1f > max=%.1f",
-                    ymd,
+            if fixed_on:
+                if not is_itm_or_atm(
+                    option_side=opt_side_hint or "",
+                    strike=pair.strike,
+                    mark_px=underlying,
+                ):
+                    logger.info(
+                        "skip expiry=%s strike=%.0f not ITM/ATM for %s mark=%.2f",
+                        ymd,
+                        pair.strike,
+                        opt_side_hint,
+                        underlying,
+                    )
+                    continue
+            else:
+                offset = atm_open_offset(pair.strike, underlying)
+                if not atm_allows_open(
                     pair.strike,
-                    offset,
-                    max_atm_off,
-                )
-                continue
+                    underlying,
+                    max_offset=max_atm_off,
+                    enabled=atm_off_on,
+                ):
+                    logger.info(
+                        "skip expiry=%s strike=%.0f atm_offset=%.1f > max=%.1f",
+                        ymd,
+                        pair.strike,
+                        offset,
+                        max_atm_off,
+                    )
+                    continue
             call_bids, call_asks, _ = self.ex.fetch_book(pair.call_inst_id, depth=5)
             put_bids, put_asks, _ = self.ex.fetch_book(pair.put_inst_id, depth=5)
             call_ask = call_asks[0].px if call_asks else None
@@ -284,12 +345,15 @@ class StrategySession:
             if put_ask is None:
                 pq = self.ex.quote(pair.put_inst_id)
                 put_ask = pq.ask if pq else None
-            sig = decide(
-                call_ask,
-                put_ask,
-                strike=pair.strike,
-                mark_px=underlying,
-            )
+            if fixed_on:
+                sig = decide_fixed(call_ask, put_ask, perp_side=fixed_perp)
+            else:
+                sig = decide(
+                    call_ask,
+                    put_ask,
+                    strike=pair.strike,
+                    mark_px=underlying,
+                )
             if sig is None:
                 continue
             opt_ask = sig.call_ask if sig.option_side == "call" else sig.put_ask
@@ -376,6 +440,13 @@ class StrategySession:
         mark = mark_px if mark_px is not None else self._mark_for_atm()
         if mark is None or mark <= 0:
             return False
+        fixed_on, fixed_perp = _fixed_direction()
+        if fixed_on:
+            opt = _option_side_for_perp(fixed_perp)
+            if not is_itm_or_atm(
+                option_side=opt, strike=float(self._pair.strike), mark_px=float(mark)
+            ):
+                return True
         return abs(float(self._pair.strike) - float(mark)) >= _ATM_DRIFT_POINTS
 
     async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None: