Merge pull request vnpy#5 from Edanflame/stream

添加tickoverview并增加流式写入参数

Author: vnpy

vnpy_postgresql/postgresql_database.py

@@ -18,6 +18,7 @@
 from vnpy.trader.database import (
     BaseDatabase,
     BarOverview,
+    TickOverview,
     DB_TZ,
     convert_tz
 )
@@ -128,16 +129,32 @@ class Meta:
         indexes = ((("symbol", "exchange", "interval"), True),)
 
 
+class DbTickOverview(Model):
+    """Tick汇总数据表映射对象"""
+
+    id: AutoField = AutoField()
+
+    symbol: str = CharField()
+    exchange: str = CharField()
+    count: int = IntegerField()
+    start: datetime = DateTimeField()
+    end: datetime = DateTimeField()
+
+    class Meta:
+        database: PeeweePostgresqlDatabase = db
+        indexes: tuple = ((("symbol", "exchange"), True),)
+
+
 class PostgresqlDatabase(BaseDatabase):
     """PostgreSQL数据库接口"""
 
     def __init__(self) -> None:
         """"""
         self.db = db
         self.db.connect()
-        self.db.create_tables([DbBarData, DbTickData, DbBarOverview])
+        self.db.create_tables([DbBarData, DbTickData, DbBarOverview, DbTickOverview])
 
-    def save_bar_data(self, bars: List[BarData]) -> bool:
+    def save_bar_data(self, bars: List[BarData], stream: bool = False) -> bool:
         """保存K线数据"""
         # 读取主键参数
         bar = bars[0]
@@ -186,6 +203,9 @@ def save_bar_data(self, bars: List[BarData]) -> bool:
             overview.start = bars[0].datetime
             overview.end = bars[-1].datetime
             overview.count = len(bars)
+        elif stream:
+            overview.end = bars[-1].datetime
+            overview.count += len(bars)
         else:
             overview.start = min(bars[0].datetime, overview.start)
             overview.end = max(bars[-1].datetime, overview.end)
@@ -201,8 +221,13 @@ def save_bar_data(self, bars: List[BarData]) -> bool:
 
         return True
 
-    def save_tick_data(self, ticks: List[TickData]) -> bool:
+    def save_tick_data(self, ticks: List[TickData], stream: bool = False) -> bool:
         """保存TICK数据"""
+        # 读取主键参数
+        tick: TickData = ticks[0]
+        symbol: str = tick.symbol
+        exchange: Exchange = tick.exchange
+
         # 将TickData数据转换为字典，并调整时区
         data = []
 
@@ -227,6 +252,34 @@ def save_tick_data(self, ticks: List[TickData]) -> bool:
                     ),
                 ).execute()
 
+        # 更新Tick汇总数据
+        overview: DbTickOverview = DbTickOverview.get_or_none(
+            DbTickOverview.symbol == symbol,
+            DbTickOverview.exchange == exchange.value,
+        )
+
+        if not overview:
+            overview: DbTickOverview = DbTickOverview()
+            overview.symbol = symbol
+            overview.exchange = exchange.value
+            overview.start = ticks[0].datetime
+            overview.end = ticks[-1].datetime
+            overview.count = len(ticks)
+        elif stream:
+            overview.end = ticks[-1].datetime
+            overview.count += len(ticks)
+        else:
+            overview.start = min(ticks[0].datetime, overview.start)
+            overview.end = max(ticks[-1].datetime, overview.end)
+
+            s: ModelSelect = DbTickData.select().where(
+                (DbTickData.symbol == symbol)
+                & (DbTickData.exchange == exchange.value)
+            )
+            overview.count = s.count()
+
+        overview.save()
+
         return True
 
     def load_bar_data(
@@ -364,6 +417,14 @@ def delete_tick_data(
             & (DbTickData.exchange == exchange.value)
         )
         count = d.execute()
+
+        # 删除Tick汇总数据
+        d2: ModelDelete = DbTickOverview.delete().where(
+            (DbTickOverview.symbol == symbol)
+            & (DbTickOverview.exchange == exchange.value)
+        )
+        d2.execute()
+
         return count
 
     def get_bar_overview(self) -> List[BarOverview]:
@@ -382,6 +443,15 @@ def get_bar_overview(self) -> List[BarOverview]:
             overviews.append(overview)
         return overviews
 
+    def get_tick_overview(self) -> List[TickOverview]:
+        """查询数据库中的Tick汇总信息"""
+        s: ModelSelect = DbTickOverview.select()
+        overviews: list = []
+        for overview in s:
+            overview.exchange = Exchange(overview.exchange)
+            overviews.append(overview)
+        return overviews
+
     def init_bar_overview(self) -> None:
         """初始化数据库中的K线汇总信息"""
         s: ModelSelect = (