[Add] query_history function for OkexGateway

vnpy/gateway/okex/okex_gateway.py

@@ -9,7 +9,7 @@ import json
 import base64
 import zlib
 from copy import copy
-from datetime import datetime
+from datetime import datetime, timedelta
 from threading import Lock
 from urllib.parse import urlencode
 
@@ -22,7 +22,8 @@ from vnpy.trader.constant import (
     Exchange,
     OrderType,
     Product,
-    Status
+    Status,
+    Interval
 )
 from vnpy.trader.gateway import BaseGateway
 from vnpy.trader.object import (
@@ -31,9 +32,11 @@ from vnpy.trader.object import (
     TradeData,
     AccountData,
     ContractData,
+    BarData,
     OrderRequest,
     CancelRequest,
     SubscribeRequest,
+    HistoryRequest
 )
 
 REST_HOST = "https://www.okex.com"
@@ -58,6 +61,18 @@ ORDERTYPE_VT2OKEX = {
 }
 ORDERTYPE_OKEX2VT = {v: k for k, v in ORDERTYPE_VT2OKEX.items()}
 
+INTERVAL_VT2OKEX = {
+    Interval.MINUTE: "60",
+    Interval.HOUR: "3600",
+    Interval.DAILY: "86400",
+}
+
+TIMEDELTA_MAP = {
+    Interval.MINUTE: timedelta(minutes=1),
+    Interval.HOUR: timedelta(hours=1),
+    Interval.DAILY: timedelta(days=1),
+}
+
 
 instruments = set()
 currencies = set()
@@ -126,6 +141,10 @@ class OkexGateway(BaseGateway):
         """"""
         pass
 
+    def query_history(self, req: HistoryRequest):
+        """"""
+        return self.rest_api.query_history(req)
+
     def close(self):
         """"""
         self.rest_api.stop()
@@ -318,6 +337,7 @@ class OkexRestApi(RestClient):
                 size=1,
                 pricetick=float(instrument_data["tick_size"]),
                 min_volume=float(instrument_data["min_size"]),
+                history_data=True,
                 gateway_name=self.gateway_name
             )
             self.gateway.on_contract(contract)
@@ -449,6 +469,75 @@ class OkexRestApi(RestClient):
             self.exception_detail(exception_type, exception_value, tb, request)
         )
 
+    def query_history(self, req: HistoryRequest):
+        """"""
+        buf = {}
+        end_time = None
+
+        for i in range(10):
+            path = f"/api/spot/v3/instruments/{req.symbol}/candles"
+
+            # Create query params
+            params = {
+                "granularity": INTERVAL_VT2OKEX[req.interval]
+            }
+
+            if end_time:
+                end = datetime.strptime(end_time, "%Y-%m-%dT%H:%M:%S.%fZ")
+                start = end - TIMEDELTA_MAP[req.interval] * 200
+
+                params["start"] = start.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+                params["end"] = end.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+            # Get response from server
+            resp = self.request(
+                "GET",
+                path,
+                params=params
+            )
+
+            # Break if request failed with other status code
+            if resp.status_code // 100 != 2:
+                msg = f"获取历史数据失败，状态码：{resp.status_code}，信息：{resp.text}"
+                self.gateway.write_log(msg)
+                break
+            else:
+                data = resp.json()
+                if not data:
+                    msg = f"获取历史数据为空"
+                    break
+
+                for l in data:
+                    ts, o, h, l, c, v = l
+                    dt = datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%fZ")
+                    bar = BarData(
+                        symbol=req.symbol,
+                        exchange=req.exchange,
+                        datetime=dt,
+                        interval=req.interval,
+                        volume=float(v),
+                        open_price=float(o),
+                        high_price=float(h),
+                        low_price=float(l),
+                        close_price=float(c),
+                        gateway_name=self.gateway_name
+                    )
+                    buf[bar.datetime] = bar
+
+                begin = data[-1][0]
+                end = data[0][0]
+                msg = f"获取历史数据成功，{req.symbol} - {req.interval.value}，{begin} - {end}"
+                self.gateway.write_log(msg)
+
+                # Update start time
+                end_time = begin
+
+        index = list(buf.keys())
+        index.sort()
+
+        history = [buf[i] for i in index]
+        return history
+
 
 class OkexWebsocketApi(WebsocketClient):
     """"""

vnpy/gateway/okexf/okexf_gateway.py

@@ -70,12 +70,6 @@ INTERVAL_VT2OKEXF = {
     Interval.DAILY: "86400",
 }
 
-TIMEDELTA_MAP = {
-    Interval.MINUTE: timedelta(minutes=1),
-    Interval.HOUR: timedelta(hours=1),
-    Interval.DAILY: timedelta(days=1),
-}
-
 
 instruments = set()
 currencies = set()
@@ -540,7 +534,10 @@ class OkexfRestApi(RestClient):
             path = f"/api/futures/v3/instruments/{req.symbol}/candles"
             
             # Create query params
-            params = {}
+            params = {
+                "granularity": INTERVAL_VT2OKEXF[req.interval]
+            }
+            
             if end_time:
                 params["end"] = end_time