feat: advance cover price set

fix: onclose cover issue
refactor: abstract strategy

Author: Yvictor

.gitignore

@@ -132,4 +132,7 @@ dmypy.json
 
 # misc
 login.json
-position.txt
+position.txt
+
+# sjtrade stratages
+sjtrade/stratages

pyproject.toml

@@ -10,6 +10,7 @@ dynamic = ["version", "description"]
 dependencies = [
     "shioaji>=0.3.4.dev7",
     "loguru",
+    "rs2py",
 ]
 requires-python = ">=3.6"
 classifiers = [

sjtrade/__init__.py

@@ -2,7 +2,7 @@
 trading with shioaji
 """
 
-__version__ = "0.3.1"
+__version__ = "0.4.0"
 
 def inject_env():
     import os

sjtrade/data.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Snapshot:
+    price: float
+    bid: float = 0
+    ask: float = 0

sjtrade/position.py

@@ -1,6 +1,5 @@
 import shioaji as sj
-
-from typing import Callable, Dict, List, Optional, Union
+from typing import List
 from threading import Lock
 from dataclasses import dataclass, field
 from shioaji.constant import (
@@ -23,7 +22,7 @@ class PositionCond:
     entry_price: List[PriceSet]
     stop_loss_price: List[PriceSet]
     stop_profit_price: List[PriceSet]
-    cover_price: List[PriceSet]
+    cover_price: List[PriceSet] = field(default_factory=list)
 
 
 @dataclass

sjtrade/simulation_shioaji.py

@@ -9,12 +9,7 @@
 import shioaji as sj
 from shioaji.constant import OrderState, Exchange, Action, TFTStockPriceType
 
-
-@dataclass
-class Snapshot:
-    price: float
-    bid: float = 0
-    ask: float = 0
+from .data import Snapshot
 
 
 class SimulationShioaji:

sjtrade/stratage.py

@@ -1,24 +1,46 @@
 import shioaji as sj
-from typing import Dict
+from typing import Dict, Optional
 from loguru import logger
-from .io.file import read_position
-from .utils import price_round
-from .position import Position, PriceSet
 from shioaji.constant import TFTStockPriceType
 
+from .io.file import read_csv_position, read_position
+from .utils import price_move, price_round, quantity_num_split
+from .position import Position, PriceSet
+from .data import Snapshot
+
 
 class StratageBase:
     name: str
 
     def entry_positions(self):
         raise NotImplementedError()
 
-    def cover_price_set(self, position: Position):
+    def cover_price_set(self, position: Position, snapshot: Optional[Snapshot] = None):
         raise NotImplementedError()
 
-    def cover_positions(self, positions: Dict[str, Position]):
+    def cover_positions(
+        self, positions: Dict[str, Position], snapshots: Dict[str, Snapshot] = dict()
+    ):
         raise NotImplementedError()
 
+    def cover_price_set_onclose(self, position: Position):
+        if position.status.open_quantity == 0:
+            return []
+        return [
+            PriceSet(
+                price=position.contract.limit_up
+                if position.status.open_quantity > 0
+                else position.contract.limit_down,
+                quantity=position.cond.quantity * -1,
+                price_type=TFTStockPriceType.LMT,
+            )
+        ]
+
+    def cover_positions_onclose(self, positions: Dict[str, Position]):
+        for code, position in positions.items():
+            position.cond.cover_price = self.cover_price_set_onclose(position)
+        return positions
+
 
 class StratageBasic(StratageBase):
     def __init__(
@@ -83,63 +105,11 @@ def entry_positions(self):
             )
         return entry_args
 
-    def cover_price_set(self, position: Position):
-        return [
-            PriceSet(
-                price=43.3,
-                quantity=position.cond.quantity * -1,
-                price_type=TFTStockPriceType.LMT,
-            )
-        ]
-
-    def cover_positions(self, positions: Dict[str, Position]):
-        for code, position in positions.items():
-            if not position.cond.cover_price:
-                position.cond.cover_price = self.cover_price_set(position)
-        return positions
+    def cover_price_set(self, position: Position, snapshot: Optional[Snapshot] = None):
+        return self.cover_price_set_onclose(position)
 
+    def cover_positions(
+        self, positions: Dict[str, Position], snapshots: Dict[str, Snapshot] = dict()
+    ):
+        return self.cover_positions_onclose()
 
-class StratageAdvance(StratageBase):
-    def __init__(
-        self,
-        entry_pct: float = -0.1,
-        stop_loss_pct: float = 0.09,
-        stop_profit_pct: float = 0.09,
-        position_filepath: str = "position.txt",
-        contracts: sj.contracts.Contracts = sj.contracts.Contracts(),
-    ) -> None:
-        self.position_filepath = position_filepath
-        self.entry_pct = entry_pct
-        self.stop_loss_pct = stop_loss_pct
-        self.stop_profit_pct = stop_profit_pct
-        self.contracts = contracts
-        self.name = "dt1"
-        self.read_position_func = read_position
-
-    def entry_positions(self):
-        positions = self.read_position_func(self.position_filepath)
-        entry_args = []
-        for code, pos in positions.items():
-            contract = self.contracts.Stocks[code]
-            if not contract:
-                logger.warning(f"Code: {code} not exist in TW Stock.")
-                continue
-            stop_loss_price = contract.reference * (
-                1 + (-1 if pos > 0 else 1) * (self.stop_loss_pct)
-            )
-            stop_profit_price = contract.reference * (
-                1 + (1 if pos > 0 else -1) * (self.stop_profit_pct)
-            )
-            entry_price = contract.reference * (
-                1 + (-1 if pos > 0 else 1) * self.entry_pct
-            )
-            entry_args.append(
-                {
-                    "code": code,
-                    "pos": pos,
-                    "entry_price": {price_round(entry_price, pos > 0): pos},
-                    "stop_profit_price": {price_round(stop_profit_price, pos < 0): pos},
-                    "stop_loss_price": {price_round(stop_loss_price, pos > 0): pos},
-                }
-            )
-        return entry_args

sjtrade/trader.py

@@ -5,6 +5,7 @@
 import shioaji as sj
 
 from .utils import quantity_split, sleep_until
+from .data import Snapshot
 from .simulation_shioaji import SimulationShioaji
 from .stratage import StratageBasic
 from .position import Position, PositionCond, PriceSet
@@ -24,6 +25,7 @@ class SJTrader:
     def __init__(self, api: sj.Shioaji, simulation: bool = False):
         self.api = api
         self.positions: Dict[str, Position] = {}
+        self.snapshots: Dict[str, Snapshot] = {}
         self._stop_loss_pct = 0.09
         self._stop_profit_pct = 0.09
         self._entry_pct = 0.05
@@ -47,6 +49,7 @@ def start(
         intraday_handler_time: datetime.time = datetime.time(8, 59, 55),
         cover_time: datetime.time = datetime.time(13, 25, 59),
     ):
+        self.set_on_tick_handler(self.update_snapshot)
         entry_future = self.executor_on_time(entry_time, self.place_entry_positions)
         self.executor_on_time(
             cancel_preorder_time,
@@ -137,6 +140,7 @@ def place_entry_order(
                     cover_price=[],
                 ),
             )
+            self.snapshots[code] = Snapshot(price=0.0)
             self.api.quote.subscribe(contract, version=QuoteVersion.v1)
             for price_set in self.positions[code].cond.entry_price:
                 price, price_quantity = price_set.price, price_set.quantity
@@ -168,6 +172,9 @@ def place_entry_positions(self) -> Dict[str, Position]:
         api.update_status()
         return self.positions
 
+    def update_snapshot(self, exchange: Exchange, tick: sj.TickSTKv1):
+        self.snapshots[tick.code].price = tick.close
+
     def cancel_preorder_handler(self, exchange: Exchange, tick: sj.TickSTKv1):
         position = self.positions[tick.code]
         if self.simulation:
@@ -229,6 +236,7 @@ def intraday_handler(self, exchange: Exchange, tick: sj.TickSTKv1):
             self.simulation_api.quote_callback(exchange, tick)
         position = self.positions[tick.code]
         self.re_entry_order(position, tick)
+        self.update_snapshot(exchange, tick)
         # 9:00 -> 13:24:49 stop loss stop profit
         self.stop_loss(position, tick)
         self.stop_profit(position, tick)
@@ -243,7 +251,7 @@ def stop_profit(self, position: Position, tick: sj.TickSTKv1):
                 cross = "under"
             for price_set in position.cond.stop_profit_price:
                 if op(tick.close, price_set.price):
-                    self.place_cover_order(position, price_set)
+                    self.place_cover_order(position, [price_set])
                     logger.info(
                         f"{position.contract.code} | price: {tick.close} cross {cross} {price_set.price} "
                         f"cover quantity: {price_set.quantity}"
@@ -259,14 +267,14 @@ def stop_loss(self, position: Position, tick: sj.TickSTKv1):
                 cross = "over"
             for price_set in position.cond.stop_profit_price:
                 if op(tick.close, price_set.price):
-                    self.place_cover_order(position, price_set)
+                    self.place_cover_order(position, [price_set])
                     logger.info(
                         f"{position.contract.code} | price: {tick.close} cross {cross} {price_set.price} "
                         f"cover quantity: {price_set.quantity}"
                     )
 
     def place_cover_order(
-        self, position: Position, price_set: Optional[PriceSet] = None
+        self, position: Position, price_sets: List[PriceSet] = []
     ):  # TODO with price quantity
         if self.simulation:
             api = self.simulation_api
@@ -275,16 +283,19 @@ def place_cover_order(
         cover_quantity = (
             position.status.open_quantity + position.status.cover_order_quantity
         )
-        if price_set is None:
-            price_set = self.stratage.cover_price_set(position)[0]
-        skip = (cover_quantity == 0)
-        if cover_quantity < 0:
+        if not price_sets:
+            price_sets = self.stratage.cover_price_set(
+                position, self.snapshots[position.contract.code]
+            )
+        if cover_quantity == 0:
+            return
+        elif cover_quantity < 0:
             action = Action.Buy
             op = max
-        elif cover_quantity > 0:
+        else:
             action = Action.Sell
             op = min
-        if not skip:
+        for price_set in price_sets:
             cover_quantity_set = op(cover_quantity, price_set.quantity)
             if cover_quantity_set:
                 quantity_s = quantity_split(cover_quantity_set, threshold=499)
@@ -305,7 +316,7 @@ def place_cover_order(
                     position.cover_trades.append(trade)
                     api.update_status(trade=trade)
 
-    def open_position_cover(self):
+    def open_position_cover(self, onclose: bool = True):
         if self.simulation:
             api = self.simulation_api
         else:
@@ -334,7 +345,12 @@ def open_position_cover(self):
                     ]:
                         api.cancel_order(trade, timeout=0)
             # event wait cancel
-        self.positions = self.stratage.cover_positions(self.positions)
+        if onclose:
+            self.positions = self.stratage.cover_positions_onclose(self.positions)
+        else:
+            self.positions = self.stratage.cover_positions(
+                self.positions, self.snapshots
+            )
         for code, position in self.positions.items():
             if position.status.open_quantity:
                 self.place_cover_order(position, position.cond.cover_price)

sjtrade/utils.py

@@ -1,6 +1,8 @@
 import math
 import time
+import rs2py
 import datetime
+from decimal import Decimal
 from typing import List, Union
 
 
@@ -28,6 +30,24 @@ def price_round(price: float, up: bool = False):
     )
 
 
+def price_move(price: float, tick: int) -> float:
+    return rs2py.get_price_tick_move(price, tick)
+
+
+def quantity_num_split(quantity: int, num: int) -> List[int]:
+    neg = 1 if quantity > 0 else -1
+    quantity_remain = abs(quantity)
+    quantity_split_res = [(quantity_remain // num) * neg] * num
+    quantity_remain -= abs(sum(quantity_split_res))
+    for idx, _ in enumerate(quantity_split_res):
+        qadd = math.ceil(quantity_remain / (num - idx))
+        quantity_remain -= qadd
+        quantity_split_res[idx] += qadd * neg
+        if not quantity_remain:
+            break
+    return quantity_split_res
+
+
 def quantity_split(quantity: float, threshold: int) -> List[int]:
     return [threshold if quantity > 0 else -threshold] * (
         abs(quantity) // threshold

tests/test_stratage.py

@@ -0,0 +1,18 @@
+import pytest
+
+from sjtrade.stratage import StratageBase
+
+
+def test_stratage_base():
+    stratage = StratageBase()
+    with pytest.raises(NotImplementedError):
+        stratage.entry_positions()
+
+    with pytest.raises(NotImplementedError):
+        stratage.cover_price_set(None)
+
+    with pytest.raises(NotImplementedError):
+        stratage.cover_positions(None)
+
+
+