feat:bybit接入

@@ -15,10 +15,11 @@ $apiInfo = new ApiInfo($key, $secret, "");
 $wsHost = "ws://bbws.keetu.com";
 $wsHost = "ws://bbws.keetu.com";
 $restHost = "http://bbapi.keetu.com";
 $restHost = "http://bbapi.keetu.com";
 $broker = new CmBroker(CmBroker::PLAT_BYBIT, $apiInfo, $wsHost, $restHost);
 $broker = new CmBroker(CmBroker::PLAT_BYBIT, $apiInfo, $wsHost, $restHost);
-// $broker->accListen(function ($data) {
-//     // var_dump($data);
-// });
+
 output("开始监听");
 output("开始监听");
-$broker->klineListen("BTCUSDT", "1m", function ($data) {
-    output($data->toArray());
+$broker->accListen(function ($data) {
+    var_dump($data);
 });
 });
+// $broker->klineListen("BTCUSDT", "1m", function ($data) {
+//     output($data->toArray());
+// });

@@ -107,6 +107,9 @@ class CmBroker
             if ($this->plat == self::PLAT_OKX) {
             if ($this->plat == self::PLAT_OKX) {
                 $this->okxAccDataHandle($data, $onData);
                 $this->okxAccDataHandle($data, $onData);
             }
             }
+            if ($this->plat == self::PLAT_BYBIT) {
+                $this->bybitAccDataHandle($data, $onData);
+            }
         });
         });
     }
     }
     //处理币安相关账户数据监听
     //处理币安相关账户数据监听
@@ -203,6 +206,41 @@ class CmBroker
         }
         }
         $this->msg("okx 无处理ws数据实现", $data);
         $this->msg("okx 无处理ws数据实现", $data);
     }
     }
+    //处理bybit账户相关数据监听
+    private function bybitAccDataHandle($data, $onData)
+    {
+        if (isset($data['topic']) && $data['topic'] == 'order') {
+            $wsDataTrade = WsDataTrade::TransferBybitTrade($data, $this->symbolInfos, function ($symbol) {
+                return $this->getSymbolSt($symbol);
+            });
+            if ($wsDataTrade != null) {
+                $wsData = new WsData($this->plat, 'trade', $trade = $wsDataTrade);
+                $onData($wsData);
+            }
+            $wsDataOrd = WsDataOrder::TransferBybitOrder($data, $this->symbolInfos, function ($symbol) {
+                return $this->getSymbolSt($symbol);
+            });
+            if ($wsDataOrd != null) {
+                $wsData = new WsData($this->plat, 'order', $trade = null, $pos = null, $order = $wsDataOrd);
+                $onData($wsData);
+            }
+            return;
+        }
+        if (isset($data['topic']) && $data['topic'] == 'position') {
+            foreach ($data['data'] as $key => $value) {
+                $wsDataPos = WsDataPos::TransferBybitPos($value, $this->symbolInfos, function ($symbol) {
+                    return $this->getSymbolSt($symbol);
+                });
+            }
+            if ($wsDataPos != null) {
+                $pos = Pos::transferWsDataPos($wsDataPos);
+                $this->positions[$wsDataPos->symbol . "_" . $wsDataPos->posSide] = $pos;
+                $wsData = new WsData($this->plat, 'pos', $trade = null, $pos = $wsDataPos);
+                $onData($wsData);
+            }
+            return;
+        }
+    }
     public function klineListen($symbol, $peroid, $onData)
     public function klineListen($symbol, $peroid, $onData)
     {
     {
         if ($this->plat == self::PLAT_BINANCE && $peroid == '1s') {
         if ($this->plat == self::PLAT_BINANCE && $peroid == '1s') {

@@ -11,6 +11,7 @@ use trader\struct\ApiInfo;
 use trader\exchange\bybit\Api as BbApi;
 use trader\exchange\bybit\Api as BbApi;
 use jytools\Websocket;
 use jytools\Websocket;
 use function jytools\output;
 use function jytools\output;
+use function jytools\getMicrotime;
 
 
 class ExBroker
 class ExBroker
 {
 {
@@ -21,6 +22,8 @@ class ExBroker
     private BbApi $api;
     private BbApi $api;
     private ?Websocket $wsAcc;
     private ?Websocket $wsAcc;
     private ?Websocket $wsKline;
     private ?Websocket $wsKline;
+    private $timerAccPing = 0;
+    private $timerKlinePing = 0;
 
 
     public function __construct(ApiInfo $apiInfo)
     public function __construct(ApiInfo $apiInfo)
     {
     {
@@ -38,10 +41,31 @@ class ExBroker
         $this->api->setHost($host);
         $this->api->setHost($host);
     }
     }
 
 
-    public function accListen(callable $onData) {}
+    public function accListen(callable $onData)
+    {
+        if (isset($this->wsAcc)) {
+            $this->wsAcc->close();
+        }
+        $this->wsAcc = new Websocket($this->host . $this->pathPri);
+        $this->wsAcc->connect(
+            function () {
+                $this->wsLogin();
+                $this->wsAccPing();
+            },
+            function ($data) use ($onData) {
+                $this->onWsDataPre($data, $onData);
+            },
+            function () {
+                swoole_timer_clear($this->timerAccPing);
+            }
+        );
+    }
 
 
     public function klineListen($symbol, $interval, callable $onData)
     public function klineListen($symbol, $interval, callable $onData)
     {
     {
+        if (isset($this->wsKline)) {
+            $this->wsKline->close();
+        }
         $this->wsKline = new Websocket($this->host . $this->pathPub);
         $this->wsKline = new Websocket($this->host . $this->pathPub);
         $this->wsKline->connect(
         $this->wsKline->connect(
             function () use ($symbol, $interval) {
             function () use ($symbol, $interval) {
@@ -51,15 +75,118 @@ class ExBroker
                         "kline.{$interval}.{$symbol}"
                         "kline.{$interval}.{$symbol}"
                     ]
                     ]
                 ]));
                 ]));
+                $this->wsKlinePing();
             },
             },
             function ($data) use ($onData) {
             function ($data) use ($onData) {
                 $data = json_decode($data, true);
                 $data = json_decode($data, true);
+                if (isset($data['op']) && $data['op'] == 'pong') {
+                    return; //pong 消息不处理
+                }
                 if (isset($data['data'])) {
                 if (isset($data['data'])) {
                     $onData($data);
                     $onData($data);
                 }
                 }
             },
             },
+            function () {
+                swoole_timer_clear($this->timerKlinePing);
+            }
         );
         );
     }
     }
+
+    private function wsKlinePing()
+    {
+        $this->timerKlinePing = swoole_timer_tick(1000 * 20, function () {
+            $pingMsg = [
+                'op' => 'ping',
+                'req_id' => (string)time(),
+            ];
+            $this->wsKline->push(json_encode($pingMsg));
+        });
+    }
+
+    private function wsAccPing()
+    {
+        $this->timerAccPing = swoole_timer_tick(1000 * 20, function () {
+            $pingMsg = [
+                'op' => 'ping',
+                'req_id' => (string)time(),
+            ];
+            $this->wsAcc->push(json_encode($pingMsg));
+        });
+    }
+
+    private function createWsSign($expires)
+    {
+        $param = "GET/realtime" . $expires;
+        return hash_hmac('sha256', $param, $this->apiInfo->secret);
+    }
+
+    private function wsLogin()
+    {
+        $expires = getMicrotime() + 10 * 1000;
+        $this->wsAcc->push(json_encode([
+            'req_id' => (string)time(),
+            'op' => 'auth',
+            'args' => [
+                $this->apiInfo->key,
+                $expires,
+                $this->createWsSign($expires)
+            ]
+        ]));
+    }
+
+    private function onWsDataPre($data, callable $onWsData)
+    {
+        $data = json_decode($data, true);
+        if (isset($data['op'])) {
+            if ($data['op'] == 'pong') {
+                return; //pong 消息不处理
+            }
+            if ($data['op'] == 'auth' && $data['success']) {
+                output('ws登录成功');
+                $this->wsSubscribe();
+                return;
+            }
+            if ($data['op'] == 'subscribe') {
+                return;
+            }
+            output('bybit ws 未处理数据', $data);
+        }
+        call_user_func($onWsData, $data);
+    }
+
+    private function wsSubscribe()
+    {
+        $this->wsAcc->push(json_encode([
+            'req_id' => (string)time(),
+            'op' => 'subscribe',
+            'args' => [
+                // 'order',
+                // 'execution',
+                'position',
+                // 'wallet'
+            ]
+        ]));
+    }
+
+    public function getKlineChannels()
+    {
+        return [
+            '1' => '1',
+            '3' => '3',
+            '5' => '5',
+            '15' => '15',
+            '30' => '30',
+            '60' => '60',
+            '120' => '120',
+            '240' => '240',
+            '360' => '360',
+            '720' => '720',
+            'D' => 'D',
+            'W' => 'W',
+            'M' => 'M'
+        ];
+    }
+
     public function getSymbolInfos()
     public function getSymbolInfos()
     {
     {
         $res = $this->api->instruments(["category" => "linear"]);
         $res = $this->api->instruments(["category" => "linear"]);
@@ -69,4 +196,95 @@ class ExBroker
         }
         }
         return $res['result']['list'];
         return $res['result']['list'];
     }
     }
+
+    public function stopListen()
+    {
+        if (isset($this->wsAcc)) {
+            $this->wsAcc->close();
+        }
+        if (isset($this->wsKline)) {
+            $this->wsKline->close();
+        }
+    }
+
+    public function placeOrder($param)
+    {
+        return $this->api->placeOrder($param);
+    }
+
+    public function getPositions($symbol = '')
+    {
+        $param = [
+            'category' => 'linear'
+        ];
+        if ($symbol) {
+            $param['symbol'] = $symbol;
+        }
+        return $this->api->getPositions($param);
+    }
+
+    public function setLeverage($symbol, $leverage)
+    {
+        $param = [
+            'category' => 'linear',
+            'symbol' => $symbol,
+            'buyLeverage' => $leverage,
+            'sellLeverage' => $leverage
+        ];
+        $res = $this->api->setLeverage($param);
+        if ($res['retCode'] == 0) {
+            output($symbol, '设置杠杆为：', $leverage);
+            return true;
+        } else {
+            output($symbol, '设置杠杆错误', $res['retMsg']);
+            return false;
+        }
+    }
+
+    public function getKlines($symbol, $interval, $limit = 200, $start = '', $end = '')
+    {
+        $param = [
+            'category' => 'linear',
+            'symbol' => $symbol,
+            'interval' => $interval,
+            'limit' => $limit
+        ];
+        if ($start) {
+            $param['start'] = $start;
+        }
+        if ($end) {
+            $param['end'] = $end;
+        }
+        return $this->api->klines($param);
+    }
+
+    public function closeAllPos()
+    {
+        $res = $this->getPositions();
+        if ($res['retCode'] == 0 && !empty($res['result']['list'])) {
+            foreach ($res['result']['list'] as $pos) {
+                if ($pos['size'] != '0') {
+                    $this->closePos($pos['symbol'], $pos['side'], $pos['size']);
+                }
+            }
+        }
+    }
+
+    private function closePos($symbol, $side, $size)
+    {
+        $param = [
+            'category' => 'linear',
+            'symbol' => $symbol,
+            'side' => $side == 'Buy' ? 'Sell' : 'Buy',
+            'orderType' => 'Market',
+            'qty' => $size
+        ];
+        $res = $this->api->placeOrder($param);
+        if ($res['retCode'] == 0) {
+            return true;
+        } else {
+            output('平仓失败', $res);
+            return false;
+        }
+    }
 }
 }

@@ -132,6 +132,31 @@ class WsDataOrder
         $status = self::getBnStatus($order['X']);
         $status = self::getBnStatus($order['X']);
         return new WsDataOrder($platform, $posSide, $symbol, $side, $price, $avgPx, $qty, $lot, $pnl, $ts, $uts, $ordType, $cliOrdId, $ordId, $status);
         return new WsDataOrder($platform, $posSide, $symbol, $side, $price, $avgPx, $qty, $lot, $pnl, $ts, $uts, $ordType, $cliOrdId, $ordId, $status);
     }
     }
+    public static function TransferBybitOrder($data, $symbolInfos, callable $toSymbolSt): WsDataOrder|null
+    {
+        $order = $data['data'];
+        $symbol = call_user_func($toSymbolSt, $order['symbol']);
+        /** @var SymbolInfo $symbolInfo */
+        $symbolInfo = $symbolInfos[$symbol] ?? null;
+        if ($symbolInfo == null) {
+            return null;
+        }
+        $platform = 'bybit';
+        $posSide = strtoupper($order['side']);
+        $side = strtoupper($order['side']);
+        $price = (float)$order['price'];
+        $avgPx = (float)$order['avg_price'];
+        $lot = (float)$order['qty'];
+        $qty = $lot;
+        $pnl = (float)$order['realised_pnl'];
+        $ts = (int)$order['transact_time'];
+        $uts = (int)$order['transact_time'];
+        $ordType = strtoupper($order['order_type']);
+        $cliOrdId = $order['order_link_id'];
+        $ordId = $order['order_id'];
+        $status = self::getBnStatus($order['order_status']);
+        return new WsDataOrder($platform, $posSide, $symbol, $side, $price, $avgPx, $qty, $lot, $pnl, $ts, $uts, $ordType, $cliOrdId, $ordId, $status);
+    }
     private static function getBnStatus(string $state)
     private static function getBnStatus(string $state)
     {
     {
         if ($state == 'NEW') {
         if ($state == 'NEW') {

@@ -71,4 +71,19 @@ class WsDataPos
         $pnl = (float)$data['up'];
         $pnl = (float)$data['up'];
         return new WsDataPos($symbol, $posSide, $qty, $lot, $avgPrice, false, $pnl);
         return new WsDataPos($symbol, $posSide, $qty, $lot, $avgPrice, false, $pnl);
     }
     }
+    public static function TransferBybitPos($data, $symbolInfos, callable $toSymbolSt): WsDataPos|null
+    {
+        $symbol = call_user_func($toSymbolSt, $data['symbol']);
+        /** @var SymbolInfo $symbolInfo */
+        $symbolInfo = $symbolInfos[$symbol] ?? null;
+        if ($symbolInfo === null) {
+            return null;
+        }
+        $posSide = strtoupper($data['side']);
+        $qty = (float)$data['size'];
+        $lot = $qty;
+        $avgPrice = (float)$data['entryPrice'];
+        $pnl = (float)$data['unrealisedPnl'];
+        return new WsDataPos($symbol, $posSide, $qty, $lot, $avgPrice, false, $pnl);
+    }
 }
 }

@@ -114,4 +114,24 @@ class WsDataTrade
         $lever = 0;
         $lever = 0;
         return new WsDataTrade($platform, $posSide, $symbol, $side, $price, $qty, $lot, $pnl, $fee, $quoteVol, $ts, $tradeId, $ordId, $cliOrdId, $lever);
         return new WsDataTrade($platform, $posSide, $symbol, $side, $price, $qty, $lot, $pnl, $fee, $quoteVol, $ts, $tradeId, $ordId, $cliOrdId, $lever);
     }
     }
+    public static function TransferBybitTrade($data, $symbolInfos, callable $toSymbolSt): WsDataTrade|null
+    {
+        $platform = 'bybit';
+        $order = $data['data'];
+        $symbol = call_user_func($toSymbolSt, $order['symbol']);
+        $posSide = strtoupper($order['side']);
+        $side = strtoupper($order['side']);
+        $price = (float)$order['price'];
+        $qty = (float)$order['qty'];
+        $lot = $qty;
+        $pnl = (float)$order['realised_pnl'];
+        $fee = (float)$order['fee'];
+        $quoteVol = $price * $qty;
+        $ts = $order['transact_time'];
+        $tradeId = $order['order_id'];
+        $ordId = $order['order_id'];
+        $cliOrdId = $order['order_link_id'];
+        $lever = 0;
+        return new WsDataTrade($platform, $posSide, $symbol, $side, $price, $qty, $lot, $pnl, $fee, $quoteVol, $ts, $tradeId, $ordId, $cliOrdId, $lever);
+    }
 }
 }