✨ feat: get_sell_limit_secs

HISTORY.md

@@ -1,6 +1,8 @@
 # History
-#
-# 2.0.0-alpha78
+
+## 2.0.0-alpha79
+*增加get_buy_limit_secs和get_sell_limit_secs接口，以查询区间内涨停个股，增加此类查询的性能。
+## 2.0.0-alpha78
 * backtest中捕获异常时，如果是TradeError类型，打印该对象自带的stack
 * Candlestick中判断峰谷时使用2倍标准差参数，以实现自适应
 * 修复当行情数据缺失时，造成的backtest迭代frame与cursor指向不一致问题

omicron/dal/influx/influxclient.py

@@ -17,7 +17,11 @@
     InfluxSchemaError,
 )
 from omicron.dal.influx.flux import Flux
-from omicron.dal.influx.serialize import DataframeSerializer, NumpySerializer
+from omicron.dal.influx.serialize import (
+    DataframeDeserializer,
+    DataframeSerializer,
+    NumpySerializer,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -454,3 +458,79 @@ async def query_org_id(self, name: str = None) -> str:
                 return org["id"]
 
         raise BadParameterError(f"can't find org with name: {name}")
+
+    async def query_reach_buy_limit(self, start: datetime.date, end:datetime.date):
+        """查询时间段[start, end]期间的涨停板
+        
+        import "math"
+        from(bucket: "zillionare")
+        |> range(start: 2024-01-04T00:00:00Z, stop: 2024-01-08T00:00:00Z)
+        |> filter(fn: (r) => r["_measurement"] == "stock_bars_1d")
+        |> filter(fn: (r) => r["_field"] == "close" or r["_field"] == "high_limit")
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> map(fn: (r) => ({r with _value: r.close - r.high_limit }))
+        |> map(fn: (r) => ({r with _value: math.abs(x: r._value) < 0.01}))
+        |> filter(fn: (r) => r._value)
+        """
+        flux = f"""
+        import "math"
+
+        from(bucket: "{self._bucket}")
+        |> range(start: {Flux.format_time(start)}, stop: {Flux.format_time(end,shift_forward=True)})
+        |> filter(fn: (r) => r["_measurement"] == "stock_bars_1d")
+        |> filter(fn: (r) => r["_field"] == "close" or r["_field"] == "high_limit")
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> map(fn: (r) => ({{r with _value: r.close - r.high_limit }}))
+        |> map(fn: (r) => ({{r with _value: math.abs(x: r._value) < 0.01}}))
+        |> filter(fn: (r) => r._value)
+        """
+
+        names = ["_", "result", "table", "_measurement","_start","_stop","_time","_value","close","code","high_limit"]
+
+        deserializer = DataframeDeserializer(
+            names=names,
+            encoding="utf-8",
+            time_col="_time",
+            engine="c",
+        )
+
+        data = await self.query(flux, deserializer)
+        return data[["code", "close", "_time"]].rename(columns={"_time": "frame"}).set_index("frame")
+
+    async def query_reach_sell_limit(self, start: datetime.date, end:datetime.date):
+        """查询时间段[start, end]期间的跌停板
+        
+        import "math"
+        from(bucket: "zillionare")
+        |> range(start: 2024-01-04T00:00:00Z, stop: 2024-01-08T00:00:00Z)
+        |> filter(fn: (r) => r["_measurement"] == "stock_bars_1d")
+        |> filter(fn: (r) => r["_field"] == "close" or r["_field"] == "low_limit")
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> map(fn: (r) => ({r with _value: r.close - r.low_limit }))
+        |> map(fn: (r) => ({r with _value: math.abs(x: r._value) < 0.01}))
+        |> filter(fn: (r) => r._value)
+        """
+        flux = f"""
+        import "math"
+
+        from(bucket: "{self._bucket}")
+        |> range(start: {Flux.format_time(start)}, stop: {Flux.format_time(end,shift_forward=True)})
+        |> filter(fn: (r) => r["_measurement"] == "stock_bars_1d")
+        |> filter(fn: (r) => r["_field"] == "close" or r["_field"] == "low_limit")
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> map(fn: (r) => ({{r with _value: r.close - r.low_limit }}))
+        |> map(fn: (r) => ({{r with _value: math.abs(x: r._value) < 0.01}}))
+        |> filter(fn: (r) => r._value)
+        """
+
+        names = ["_", "result", "table", "_measurement","_start","_stop","_time","_value","close","code","low_limit"]
+
+        deserializer = DataframeDeserializer(
+            names=names,
+            encoding="utf-8",
+            time_col="_time",
+            engine="c",
+        )
+
+        data = await self.query(flux, deserializer)
+        return data[["code", "close", "_time"]].rename(columns={"_time": "frame"}).set_index("frame")

omicron/models/stock.py

@@ -2,7 +2,7 @@
 import itertools
 import logging
 import re
-from typing import Dict, Generator, Iterable, List, Tuple, Union
+from typing import Dict, Generator, Iterable, List, Optional, Tuple, Union
 
 import arrow
 import cfg4py
@@ -1460,3 +1460,39 @@ async def get_latest_price(cls, codes: Iterable[str]) -> List[str]:
             else:
                 _converted_data.append(float(_data))
         return _converted_data
+
+    @classmethod
+    async def get_buy_limit_secs(cls, start: datetime.date, end: Optional[datetime.date]=None)->pd.DataFrame:
+        """查询在[start, end]区间涨停的个股代码
+        
+            如果只查询某一天数据，则可省略end。本方法查询出来的，有可能有很小的概率，因为浮点数舍入误差出现不准确的情况。
+            Args:
+                start: 起始时间，包含
+                end: 截止时间，包含
+            Returns:
+                区间内涨停列表，一个包含日期、code和收盘价的dataframe
+        """
+        end = end or start
+
+        client = get_influx_client()
+        data = await client.query_reach_buy_limit(start, end)
+
+        return data
+
+    @classmethod
+    async def get_sell_limit_secs(cls, start: datetime.date, end: Optional[datetime.date]=None)->pd.DataFrame:
+        """查询在[start, end]区间跌停的个股代码
+        
+            如果只查询某一天数据，则可省略end，本方法查询出来的，有可能有很小的概率，因为浮点数舍入误差出现不准确的情况。
+            Args:
+                start: 起始时间，包含
+                end: 截止时间，包含
+            Returns:
+                区间内涨停列表，一个包含日期、code和收盘价的dataframe
+        """
+        end = end or start
+
+        client = get_influx_client()
+        data = await client.query_reach_sell_limit(start, end)
+
+        return data

pyproject.toml

@@ -11,7 +11,7 @@ name = "zillionare-omicron"
 packages = [
     {include = "omicron"}
 ]
-version = "2.0.0a78"
+version = "2.0.0a79"
 description = "Core Library for Zillionare"
 authors = ["jieyu <code@jieyu.ai>"]
 license = "MIT"