Create api/kucoin.py

api/kucoin.py

@@ -0,0 +1,50 @@
+import time
+import ccxt
+from typing import Optional, Dict, List, Any
+from config import SessionManager
+from models.market_data import MarketData, SymbolInfo
+
+
+class KuCoinClient:
+    def __init__(self):
+        s = SessionManager()
+        self.client = ccxt.kucoinfutures(
+            {
+                "apiKey": s.kucoin_key,
+                "secret": s.kucoin_secret,
+                "password": s.kucoin_passphrase,
+            }
+        )
+
+    def get_kline_data(
+        self,
+        symbol: str,
+        granularity: int,
+        from_time: int,
+        to_time: Optional[int] = None,
+    ) -> MarketData:
+        if to_time is None:
+            to_time = int(time.time() * 1000)
+        data = self.client.fetch_ohlcv(symbol, f"{granularity}m", since=from_time)
+        formatted = [
+            [d[0], str(d[1]), str(d[2]), str(d[3]), str(d[4]), str(d[5])] for d in data
+        ]
+        return MarketData.from_kucoin_response({"data": formatted}, symbol, granularity)
+
+    def get_ticker(self, symbol: str) -> Dict[str, Any]:
+        t = self.client.fetch_ticker(symbol)
+        return {"price": str(t["last"]), "time": t["timestamp"]}
+
+    def get_symbols_list(self) -> List[SymbolInfo]:
+        return [
+            SymbolInfo.from_kucoin_data(
+                {
+                    "symbol": m["id"],
+                    "name": m["symbol"],
+                    "baseCurrency": m["base"],
+                    "quoteCurrency": m["quote"],
+                    "maxLeverage": 20,
+                }
+            )
+            for m in self.client.fetch_markets()
+        ]