952 lines
36 KiB
PHP
952 lines
36 KiB
PHP
<?php
|
|
|
|
namespace App\Utils\Workerman;
|
|
|
|
use App\Jobs\UpdateCurrencyPrice;
|
|
use Workerman\Connection\AsyncTcpConnection;
|
|
use Workerman\Lib\Timer;
|
|
use Illuminate\Support\Facades\DB;
|
|
use App\{Currency, CurrencyContact,CurrencyMatch, CurrencyQuotation, MarketHour, UserChat};
|
|
use App\Jobs\{CoinTradeHandel, EsearchMarket, LeverUpdate, LeverPushPrice, SendMarket, WriteMarket, HandleMicroTrade};
|
|
use Illuminate\Support\Carbon;
|
|
use Illuminate\Support\Facades\Redis;
|
|
|
|
class WsConnection
|
|
{
|
|
protected $server_address = 'ws://api.huobi.pro:443/ws';
|
|
//protected $server_address = 'ws://api.huobi.br.com:443/'; //ws国内开发调试
|
|
protected $server_ping_freq = 5; //服务器ping检测周期,单位秒
|
|
protected $server_time_out = 2; //服务器响应超时
|
|
protected $send_freq = 2; //写入和发送数据的周期，单位秒
|
|
protected $micro_trade_freq = 1; //秒合约处理时间周期
|
|
|
|
protected $worker_id;
|
|
|
|
protected $events = [
|
|
'onConnect',
|
|
'onClose',
|
|
'onMessage',
|
|
'onError',
|
|
'onBufferFull',
|
|
'onBufferDrain',
|
|
];
|
|
|
|
protected static $marketKlineData = [];
|
|
protected static $marketDepthData = [];
|
|
protected static $matchTradeData = []; //撮合交易全站交易
|
|
|
|
protected $connection;
|
|
|
|
protected $timer;
|
|
|
|
protected $pingTimer;
|
|
|
|
protected $sendKlineTimer;
|
|
|
|
protected $sendDepthTimer;
|
|
|
|
protected $depthTimer;
|
|
|
|
protected $handleTimer;
|
|
|
|
protected $microTradeHandleTimer;
|
|
|
|
protected $sendMatchTradeTimer;
|
|
|
|
protected $subscribed = [];
|
|
|
|
protected $topicTemplate = [
|
|
'sub' => [
|
|
'market_kline' => 'market.$symbol.kline.$period',
|
|
'market_detail' => 'market.$symbol.detail',
|
|
'market_depth' => 'market.$symbol.depth.$type',
|
|
'market_trade' => 'market.$symbol.trade.detail', //成交的交易
|
|
// 'market_ticker'=>'market.$symbol.ticker',//最新价跳动
|
|
],
|
|
];
|
|
|
|
public function __construct($worker_id)
|
|
{
|
|
$this->worker_id = $worker_id;
|
|
AsyncTcpConnection::$defaultMaxPackageSize = 1048576000;
|
|
}
|
|
|
|
public function getConnection()
|
|
{
|
|
return $this->connection;
|
|
}
|
|
|
|
/**
|
|
* 绑定所有事件到连接
|
|
*
|
|
* @return void
|
|
*/
|
|
protected function bindEvent()
|
|
{
|
|
foreach ($this->events as $key => $event) {
|
|
if (method_exists($this, $event)) {
|
|
$this->connection && $this->connection->$event = [$this, $event];
|
|
//echo '绑定' . $event . '事件成功' . PHP_EOL;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 解除连接所有绑定事件
|
|
*
|
|
* @return void
|
|
*/
|
|
protected function unBindEvent()
|
|
{
|
|
foreach ($this->events as $key => $event) {
|
|
if (method_exists($this, $event)) {
|
|
$this->connection && $this->connection->$event = null;
|
|
//echo '解绑' . $event . '事件成功' . PHP_EOL;
|
|
}
|
|
}
|
|
}
|
|
|
|
public function getSubscribed($topic = null)
|
|
{
|
|
if (is_null($topic)) {
|
|
return $this->subscribed;
|
|
}
|
|
return $this->subscribed[$topic] ?? null;
|
|
}
|
|
|
|
protected function setSubscribed($topic, $value)
|
|
{
|
|
$this->subscribed[$topic] = $value;
|
|
}
|
|
|
|
protected function delSubscribed($topic)
|
|
{
|
|
unset($this->subscribed[$topic]);
|
|
}
|
|
|
|
public function connect()
|
|
{
|
|
$this->connection = new AsyncTcpConnection($this->server_address);
|
|
$this->bindEvent();
|
|
$this->connection->transport = 'ssl';
|
|
$this->connection->connect();
|
|
}
|
|
|
|
public function onConnect($con)
|
|
{
|
|
//连接成功后定期发送ping数据包检测服务器是否在线
|
|
$this->timer = Timer::add($this->server_ping_freq, [$this, 'ping'], [$this->connection], true);
|
|
if ($this->worker_id < 8) {
|
|
$this->sendKlineTimer = Timer::add($this->send_freq, [$this, 'writeMarketKline'], [], true);
|
|
} else {
|
|
$this->depthTimer = Timer::add($this->send_freq, [$this, 'sendDepthData'], [], true);
|
|
$this->sendMatchTradeTimer = Timer::add($this->send_freq, [$this, 'sendMatchTradeData'], [], true);
|
|
}
|
|
|
|
if ($this->worker_id == 5) {
|
|
$this->handleTimer = Timer::add($this->send_freq, [self::class, 'sendLeverHandle'], [], true);
|
|
}
|
|
|
|
if ($this->worker_id == 0) {
|
|
$this->microTradeHandleTimer = Timer::add($this->micro_trade_freq, [self::class, 'handleMicroTrade'], [], true);
|
|
}
|
|
//添加订阅事件代码
|
|
$this->subscribe($con);
|
|
}
|
|
|
|
public function onClose($con)
|
|
{
|
|
echo $this->server_address . '连接关闭' . PHP_EOL;
|
|
$path = base_path() . '/storage/logs/wss/';
|
|
$filename = date('Ymd') . '.log';
|
|
file_exists($path) || @mkdir($path);
|
|
error_log(date('Y-m-d H:i:s') . ' ' . $this->server_address . '连接关闭' . PHP_EOL, 3, $path . $filename);
|
|
//解除事件
|
|
$this->timer && Timer::del($this->timer);
|
|
$this->sendKlineTimer && Timer::del($this->sendKlineTimer);
|
|
$this->pingTimer && Timer::del($this->pingTimer);
|
|
$this->depthTimer && Timer::del($this->depthTimer);
|
|
$this->handleTimer && Timer::del($this->handleTimer);
|
|
$this->microTradeHandleTimer && Timer::del($this->microTradeHandleTimer);
|
|
$this->unBindEvent();
|
|
unset($this->connection);
|
|
$this->connection = null;
|
|
$this->subscribed = null; //清空订阅
|
|
echo '尝试重新连接' . PHP_EOL;
|
|
$this->connect();
|
|
}
|
|
|
|
public function close($msg)
|
|
{
|
|
$path = base_path() . '/storage/logs/wss/';
|
|
$filename = date('Ymd') . '.log';
|
|
file_exists($path) || @mkdir($path);
|
|
error_log(date('Y-m-d H:i:s') . ' ' . $msg, 3, $path . $filename);
|
|
$this->connection->destroy();
|
|
}
|
|
|
|
protected function makeSubscribeTopic($topic_template, $param)
|
|
{
|
|
$need_param = [];
|
|
$match_count = preg_match_all('/\$([a-zA-Z_]\w*)/', $topic_template, $need_param);
|
|
if ($match_count > 0 && count(reset($need_param)) > count($param)) {
|
|
throw new \Exception('所需参数不匹配');
|
|
}
|
|
$diff = array_diff(next($need_param), array_keys($param));
|
|
if (count($diff) > 0) {
|
|
throw new \Exception('topic:' . $topic_template . '缺少参数：' . implode(',', $diff));
|
|
}
|
|
return preg_replace_callback('/\$([a-zA-Z_]\w*)/', function ($matches) use ($param) {
|
|
extract($param);
|
|
$value = $matches[1];
|
|
return $$value ?? '';
|
|
}, $topic_template);
|
|
}
|
|
|
|
public function onBufferFull()
|
|
{
|
|
echo 'buffer is full' . PHP_EOL;
|
|
}
|
|
|
|
protected function subscribe($con)
|
|
{
|
|
$periods = ['1min', '5min', '15min', '30min', '60min', '1day', '1mon', '1week']; //['1day', '1min'];
|
|
if ($this->worker_id < 8) {
|
|
$value = $periods[$this->worker_id];
|
|
echo '进程'. $this->worker_id . '开始订阅' . $value . '数据' . PHP_EOL;
|
|
$this->subscribeKline($con, $value); //订阅k线行情
|
|
} else {
|
|
if ($this->worker_id == 8) {
|
|
$this->subscribeMarketDepth($con); //订阅盘口数据
|
|
$this->subscribeMarketTrade($con); //订阅全站交易数据
|
|
}
|
|
}
|
|
}
|
|
|
|
//订阅回调
|
|
protected function onSubscribe($data)
|
|
{
|
|
if ($data->status == 'ok') {
|
|
echo $data->subbed . '订阅成功' . PHP_EOL;
|
|
} else {
|
|
echo '订阅失败:' . $data->{'err-msg'} . PHP_EOL;
|
|
}
|
|
}
|
|
/**
|
|
* 订阅全站交易(已完成)
|
|
* @param \Workerman\Connection\ConnectionInterface $con
|
|
* @param \App\Models\CurrencyMatch $currency_match
|
|
* @return void
|
|
*/
|
|
public function subscribeMarketTrade($con)
|
|
{
|
|
$currency_list = CurrencyMatch::getHuobiMatchs();
|
|
foreach ($currency_list as $key => $currency_match) {
|
|
$param = [
|
|
'symbol' => $currency_match->match_name,
|
|
];
|
|
$topic = $this->makeSubscribeTopic($this->topicTemplate['sub']['market_trade'], $param);
|
|
$sub_data = json_encode([
|
|
'sub' => $topic,
|
|
'id' => $topic,
|
|
]);
|
|
$subscribed_data = $this->getSubscribed($topic);
|
|
$match_data = $subscribed_data['match'] ?? [];
|
|
$match_data[] = $currency_match;
|
|
$this->setSubscribed($topic, [
|
|
'callback' => 'onMatchTrade',
|
|
'match' => $match_data,
|
|
]);
|
|
// 未订阅过的才能订阅
|
|
if (is_null($subscribed_data)) {
|
|
$con->send($sub_data);
|
|
}
|
|
}
|
|
}
|
|
|
|
//订阅K线行情
|
|
protected function subscribeKline($con, $period)
|
|
{
|
|
$currency_match = CurrencyMatch::getHuobiMatchs();
|
|
foreach ($currency_match as $key => $value) {
|
|
$param = [
|
|
'symbol' => $value->match_name,
|
|
'period' => $period,
|
|
];
|
|
$topic = $this->makeSubscribeTopic($this->topicTemplate['sub']['market_kline'], $param);
|
|
$sub_data = json_encode([
|
|
'sub' => $topic,
|
|
'id' => $topic,
|
|
//'freq-ms' => 5000, //推送频率，实测只能是0和5000，与官网文档不符
|
|
]);
|
|
//未订阅过的才能订阅
|
|
if (is_null($this->getSubscribed($topic))) {
|
|
$this->setSubscribed($topic, [
|
|
'callback' => 'onMarketKline',
|
|
'match' => $value
|
|
]);
|
|
$con->send($sub_data);
|
|
}
|
|
}
|
|
}
|
|
|
|
protected function onMarketKline($con, $data, $match)
|
|
{
|
|
$topic = $data->ch;
|
|
$msg = date('Y-m-d H:i:s') . ' 进程' . $this->worker_id . '接收' . $topic . '行情' . PHP_EOL;
|
|
list($name, $symbol, $detail_name, $period) = explode('.', $topic);
|
|
$subscribed_data = $this->getSubscribed($topic);
|
|
$currency_match = $subscribed_data['match'];
|
|
$key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
// $currency_match->market_from == 2
|
|
$tick = $data->tick;
|
|
|
|
$currencyInfo=Currency::where('name',$currency_match->currency_name)->first();
|
|
$sisValue =$currencyInfo->oncontact;
|
|
|
|
|
|
$market_data = [
|
|
'id' => $tick->id,
|
|
'period' => $period,
|
|
'base-currency' => $currency_match->currency_name,
|
|
'quote-currency' => $currency_match->legal_name,
|
|
'open' => sctonum($tick->open),
|
|
'close' => sctonum($tick->close),
|
|
'high' => sctonum($tick->high),
|
|
'low' => sctonum($tick->low),
|
|
'vol' => sctonum($tick->vol),
|
|
'amount' => sctonum($tick->amount),
|
|
];
|
|
|
|
$kline_data = [
|
|
'type' => 'kline',
|
|
'period' => $period,
|
|
'match_id' => $currency_match->id,
|
|
'currency_id' => $currency_match->currency_id,
|
|
'currency_name' => $currency_match->currency_name,
|
|
'legal_id' => $currency_match->legal_id,
|
|
'legal_name' => $currency_match->legal_name,
|
|
'open' => sctonum($tick->open),
|
|
'close' => sctonum($tick->close),//实时价格
|
|
'high' => sctonum($tick->high),
|
|
'low' => sctonum($tick->low),
|
|
'symbol' => $currency_match->currency_name . '/' . $currency_match->legal_name,
|
|
'volume' => sctonum($tick->amount),
|
|
'time' => $tick->id * 1000,
|
|
];
|
|
|
|
$forward_sj = CurrencyMatch::forward_sj();
|
|
$forward_data = [
|
|
'type' => 'kline',
|
|
'period' => '1min',
|
|
'match_id' => $forward_sj['currency_id'],
|
|
'currency_id' => $forward_sj['currency_id'],
|
|
'currency_name' => $forward_sj['name'],
|
|
'legal_id' => 3,
|
|
'legal_name' => 'USDT',
|
|
'open' => $forward_sj['open'],
|
|
'close' => $forward_sj['close'],//实时价格
|
|
'high' => $forward_sj['high'],
|
|
'low' => $forward_sj['low'],
|
|
'symbol' => $forward_sj['name'] . '/' . 'USDT',
|
|
'volume' => $forward_sj['volume'],
|
|
'time' => time().'000',
|
|
];
|
|
|
|
|
|
|
|
|
|
if($period=='1week'){
|
|
$kline_data = $forward_data;
|
|
$key = $forward_sj['name'].'.USDT';
|
|
|
|
|
|
}
|
|
|
|
|
|
$foreign_sj = CurrencyMatch::foreign_sj();
|
|
$foreign_data = [
|
|
'type' => 'kline',
|
|
//'period' => '1dy',
|
|
'period' => '1min',
|
|
'match_id' => $foreign_sj['currency_id'],
|
|
'currency_id' => $foreign_sj['currency_id'],
|
|
'currency_name' => $foreign_sj['name'],
|
|
'legal_id' => 3,
|
|
'legal_name' => 'USDT',
|
|
'open' => $foreign_sj['open'],
|
|
'close' => $foreign_sj['close'],//实时价格
|
|
'high' => $foreign_sj['high'],
|
|
'low' => $foreign_sj['low'],
|
|
'symbol' => $foreign_sj['name'] . '/' . 'USDT',
|
|
'volume' => $foreign_sj['volume'],
|
|
'time' => time().'000',
|
|
];
|
|
if($period=='1mon'){
|
|
$kline_data = $foreign_data;
|
|
$key = $foreign_sj['name'].'.USDT';
|
|
|
|
}
|
|
|
|
//EsearchMarket::dispatch($market_data)->onQueue('esearch:market:' . $period);
|
|
// $key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
|
|
|
|
self::$marketKlineData[$period][$key] = [
|
|
'market_data' => $market_data,
|
|
'kline_data' => $kline_data,
|
|
'forward_data' => $forward_data,
|
|
'foreign_data' => $foreign_data,
|
|
];
|
|
|
|
$rand = rand(1,4);
|
|
if($period=='1week'&&$rand==2){
|
|
//推送币种的日行情(带涨副)
|
|
$change = $this->calcIncreasePair($forward_data);
|
|
bc_comp($change, 0) > 0 && $change = '+' . $change;
|
|
//追加涨副等信息
|
|
$daymarket_data = [
|
|
'type' => 'daymarket',
|
|
'period' => '1dy', //1dy
|
|
'change' => $change,
|
|
'now_price' => $forward_data['close'],
|
|
'api_form' => 'sina_api',
|
|
];
|
|
|
|
|
|
|
|
$kline_data = array_merge($forward_data, $daymarket_data);
|
|
|
|
|
|
self::$marketKlineData[$period][$key]['kline_data'] = $kline_data;
|
|
|
|
|
|
}
|
|
|
|
|
|
if($period=='1mon'&&$rand==2){
|
|
//推送币种的日行情(带涨副)
|
|
$change = $this->calcIncreasePair($foreign_data);
|
|
bc_comp($change, 0) > 0 && $change = '+' . $change;
|
|
//追加涨副等信息
|
|
$daymarket_data = [
|
|
'type' => 'daymarket',
|
|
'period' => '1dy', //1dy
|
|
'change' => $change,
|
|
'now_price' => $foreign_data['close'],
|
|
'api_form' => 'sina_api',
|
|
];
|
|
|
|
|
|
|
|
$kline_data = array_merge($foreign_data, $daymarket_data);
|
|
|
|
|
|
self::$marketKlineData[$period][$key]['kline_data'] = $kline_data;
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if ($period == '1day') {
|
|
//推送币种的日行情(带涨副)
|
|
$change = $this->calcIncreasePair($kline_data);
|
|
bc_comp($change, 0) > 0 && $change = '+' . $change;
|
|
//追加涨副等信息
|
|
$daymarket_data = [
|
|
'type' => 'daymarket',
|
|
'change' => $change,
|
|
'now_price' => $market_data['close'],
|
|
'api_form' => 'huobi_websocket',
|
|
];
|
|
|
|
|
|
$kline_data = array_merge($kline_data, $daymarket_data);
|
|
|
|
$currencyInfo=Currency::where('name',$currency_match->currency_name)->first();
|
|
if($currencyInfo->enabled==1){
|
|
// $sis = bcmul($currencyInfo->yujicontact,$currencyInfo->min_unit/100,4);
|
|
$das=CurrencyContact::where('currency_id',$currencyInfo->id)->where('time',date('YmdHi',time()))->first();
|
|
if($currencyInfo->yujicontact !=$currencyInfo->min_unit && !$das){
|
|
$sis=bcmul($kline_data['open'],$currencyInfo->count/$currencyInfo->min_unit/100,4);
|
|
$yujicontact=$currencyInfo->yujicontact + 1;
|
|
}else{
|
|
$sis=0;
|
|
$yujicontact=$currencyInfo->yujicontact;
|
|
}
|
|
|
|
$oncontact=bcadd($currencyInfo->oncontact,$sis,4);
|
|
|
|
|
|
// if($currencyInfo->yujicontact <= 0 && $oncontact <=$currencyInfo->yujicontact){
|
|
// $oncontact=$currencyInfo->yujicontact;
|
|
// }
|
|
|
|
// if($currencyInfo->yujicontact > 0 && $oncontact >=$currencyInfo->yujicontact){
|
|
// $oncontact=$currencyInfo->yujicontact;
|
|
// }
|
|
|
|
Currency::where(['name' => $currency_match->currency_name])->update(['oncontact'=>$oncontact,'yujicontact'=>$yujicontact]);
|
|
|
|
$rsd=[
|
|
'currency_id'=>$currencyInfo->id,
|
|
'time'=>date('YmdHi',time()),
|
|
'oncontact'=>$oncontact
|
|
];
|
|
if(!$das){
|
|
DB::table('currency_contact')->insert($rsd);
|
|
}
|
|
|
|
}
|
|
|
|
$sisValue =$currencyInfo->oncontact;
|
|
if($sisValue != 0){
|
|
// $kline_data['high'] = round($kline_data['high'] + $sisValue,8);
|
|
// $kline_data['close'] = round($kline_data['close'] + $sisValue,8);
|
|
// $kline_data['low'] = round($kline_data['low'] + $sisValue,8);
|
|
// $kline_data['open'] = round($kline_data['open'] + $sisValue,8);
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self::$marketKlineData[$period][$key]['kline_data'] = $kline_data;
|
|
//SendMarket::dispatch($kline_data)->onQueue('kline.1day');
|
|
/*
|
|
//存入数据库
|
|
CurrencyQuotation::getInstance($currency_match->legal_id, $currency_match->currency_id)
|
|
->updateData([
|
|
'change' => $change,
|
|
'now_price' => $tick->close,
|
|
'volume' => $tick->amount,
|
|
]);
|
|
*/
|
|
} /*elseif ($period == '5min') {
|
|
echo str_repeat('*', 80) . PHP_EOL;
|
|
dump($kline_data);
|
|
echo str_repeat('*', 80) . PHP_EOL;
|
|
}*/
|
|
}
|
|
|
|
//订阅盘口数据
|
|
protected function subscribeMarketDepth($con)
|
|
{
|
|
$currency_match = CurrencyMatch::getHuobiMatchs();
|
|
foreach ($currency_match as $key => $value) {
|
|
$param = [
|
|
'symbol' => $value->match_name,
|
|
'type' => 'step0',
|
|
];
|
|
$topic = $this->makeSubscribeTopic($this->topicTemplate['sub']['market_depth'], $param);
|
|
$sub_data = json_encode([
|
|
'sub' => $topic,
|
|
'id' => $topic,
|
|
]);
|
|
//未订阅过的才能订阅
|
|
if (is_null($this->getSubscribed($topic))) {
|
|
$this->setSubscribed($topic, [
|
|
'callback' => 'onMarketDepth',
|
|
'match' => $value
|
|
]);
|
|
$con->send($sub_data);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 撮合交易全站交易数据回调
|
|
* @param \Workerman\Connection\ConnectionInterface $con
|
|
* @param array $data
|
|
* @param \Illuminate\Database\Eloquent\Collection $currency_matches
|
|
* @return void
|
|
*/
|
|
protected function onMatchTrade($con, $data, $currency_matches)
|
|
{
|
|
$topic = $data->ch;
|
|
$data = $data->tick->data;
|
|
foreach ($currency_matches as $key => $currency_match) {
|
|
$symbol_key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
$trade_data = [
|
|
'type' => 'match_trade',
|
|
'symbol' => $currency_match->currency_name . '/' . $currency_match->legal_name,
|
|
'base-currency' => $currency_match->currency_name,
|
|
'quote-currency' => $currency_match->legal_name,
|
|
'currency_id' => $currency_match->currency_id,
|
|
'currency_name' => $currency_match->currency_name,
|
|
'legal_id' => $currency_match->legal_id,
|
|
'legal_name' => $currency_match->legal_name,
|
|
'data' => $data,
|
|
];
|
|
self::$matchTradeData[$symbol_key] = $trade_data;
|
|
}
|
|
}
|
|
|
|
// protected function onMarketDepth($con, $data, $currency_matches)
|
|
// {
|
|
// try {
|
|
// $topic = $data->ch;
|
|
// $limit = 10;
|
|
// $tick = $data->tick;
|
|
// $bids = array_slice($tick->bids, 0, $limit);
|
|
// $asks = array_slice($tick->asks, 0, $limit);
|
|
// krsort($asks);
|
|
// $asks = array_values($asks);
|
|
// // 将模拟克隆的交易对也添加上数据
|
|
// foreach ($currency_matches as $key => $currency_match) {
|
|
// $depth_data = [
|
|
// 'type' => 'market_depth',
|
|
// 'symbol' => $currency_match->currency_name . '/' . $currency_match->legal_name,
|
|
// 'base-currency' => $currency_match->currency_name,
|
|
// 'quote-currency' => $currency_match->legal_name,
|
|
// 'currency_id' => $currency_match->currency_id,
|
|
// 'currency_name' => $currency_match->currency_name,
|
|
// 'legal_id' => $currency_match->legal_id,
|
|
// 'legal_name' => $currency_match->legal_name,
|
|
// 'bids' => $bids, //买入盘口
|
|
// 'asks' => $asks, //卖出盘口
|
|
// ];
|
|
// $symbol_key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
// self::$marketDepthData[$symbol_key] = $depth_data;
|
|
// }
|
|
// } catch (\Throwable $th) {
|
|
|
|
// }
|
|
// }
|
|
|
|
//盘口数据回调
|
|
protected function onMarketDepth($con, $data, $match)
|
|
{
|
|
$topic = $data->ch;
|
|
$subscribed_data = $this->getSubscribed($topic);
|
|
$currency_match = $subscribed_data['match'];
|
|
krsort($data->tick->asks);
|
|
$data->tick->asks = array_values($data->tick->asks);
|
|
|
|
$rand = rand(1,3);
|
|
if($rand==1){//期货
|
|
$info = CurrencyMatch::forward_sj();
|
|
}else if($rand==2){
|
|
$info = CurrencyMatch::foreign_sj();
|
|
}
|
|
|
|
if($rand==1||$rand==2){//期货 或外汇
|
|
|
|
$array = array(1,1,1,1,1,1,1,1,1,1);
|
|
foreach ($array as $k=>$v){
|
|
$bids[$k][0] = rand(1,2)==1?$info['close']+$info['close']/rand(2,6):$info['close']-$info['close']/rand(30,50);
|
|
$bids[$k][1] = '0.'.rand(0,10);
|
|
}
|
|
foreach ($array as $k=>$v){
|
|
$asks[$k][0] = rand(1,2)==1?$info['close']+$info['close']/rand(2,6):$info['close']-$info['close']/rand(30,50);
|
|
$asks[$k][1] = '0.'.rand(0,10);
|
|
}
|
|
|
|
$depth_data = [
|
|
'type' => 'market_depth',
|
|
'symbol' => $info['name'] . '/' . 'USDT',
|
|
'base-currency' => $info['name'],//币名称
|
|
'quote-currency' => 'USDT',//换汇名称
|
|
'currency_id' => $info['currency_id'],//币id
|
|
'currency_name' => $info['name'],//币名称
|
|
'legal_id' => 3,
|
|
'legal_name' => 'USDT',
|
|
'bids' => $bids, //买入盘口
|
|
'asks' => $asks, //卖出盘口
|
|
];
|
|
|
|
}else if($rand==3){//元数据
|
|
$depth_data = [
|
|
'type' => 'market_depth',
|
|
'symbol' => $currency_match->currency_name . '/' . $currency_match->legal_name,
|
|
'base-currency' => $currency_match->currency_name,
|
|
'quote-currency' => $currency_match->legal_name,
|
|
'currency_id' => $currency_match->currency_id,
|
|
'currency_name' => $currency_match->currency_name,
|
|
'legal_id' => $currency_match->legal_id,
|
|
'legal_name' => $currency_match->legal_name,
|
|
'bids' => array_slice($data->tick->bids, 0, 10), //买入盘口
|
|
'asks' => array_slice($data->tick->asks, 0, 10), //卖出盘口
|
|
];
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
$symbol_key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
self::$marketDepthData[$symbol_key] = $depth_data;
|
|
}
|
|
|
|
/**
|
|
* 发送盘口数据
|
|
*
|
|
* @return void
|
|
*/
|
|
|
|
public function sendDepthData()
|
|
{
|
|
$market_depth = self::$marketDepthData;
|
|
foreach ($market_depth as $depth_data) {
|
|
$currencyInfo=Currency::where('name',$depth_data['base-currency'])->first();
|
|
if($currencyInfo){
|
|
$sisValue =$currencyInfo->oncontact;
|
|
if($sisValue>0){
|
|
foreach($depth_data['asks'] as $k => $asks){
|
|
$depth_data['asks'][$k][0] = round($asks[0] + $sisValue,8);
|
|
}
|
|
foreach($depth_data['bids'] as $k => $bids){
|
|
$depth_data['bids'][$k][0] = round($bids[0] + $sisValue,8);
|
|
}
|
|
}
|
|
}
|
|
SendMarket::dispatch($depth_data)->onQueue('market.depth');
|
|
}
|
|
}
|
|
|
|
public function sendDepthData111()
|
|
{
|
|
$market_depth = self::$marketDepthData;
|
|
foreach ($market_depth as $depth_data) {
|
|
SendMarket::dispatch($depth_data)->onQueue('market.depth');
|
|
}
|
|
}
|
|
|
|
//取消订阅
|
|
protected function unsubscribe()
|
|
{
|
|
}
|
|
|
|
protected function onUnsubscribe()
|
|
{
|
|
}
|
|
|
|
public function onMessage($con, $data)
|
|
{
|
|
$data = gzdecode($data);
|
|
$data = json_decode($data, false, 512, JSON_BIGINT_AS_STRING);
|
|
|
|
if (isset($data->ping)) {
|
|
$this->onPong($con, $data);
|
|
} elseif (isset($data->pong)) {
|
|
$this->onPing($con, $data);
|
|
} elseif (isset($data->id) && $this->getSubscribed($data->id) != null) {
|
|
$this->onSubscribe($data);
|
|
} elseif (isset($data->id)) {
|
|
|
|
} else {
|
|
$this->onData($con, $data);
|
|
}
|
|
}
|
|
|
|
protected function onData($con, $data)
|
|
{
|
|
if (isset($data->ch)) {
|
|
$subscribed = $this->getSubscribed($data->ch);
|
|
if ($subscribed != null) {
|
|
//调用回调处理
|
|
$callback = $subscribed['callback'];
|
|
$this->$callback($con, $data, $subscribed['match']);
|
|
} else {
|
|
//不在订阅中的数据
|
|
}
|
|
} else {
|
|
echo '未知数据' . PHP_EOL;
|
|
var_dump($data);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 发送全站交易数据
|
|
*
|
|
* @return void
|
|
*/
|
|
public function sendMatchTradeData()
|
|
{
|
|
$market_trade = self::$matchTradeData;
|
|
foreach ($market_trade as $trade_data) {
|
|
SendMarket::dispatch($trade_data)->onQueue('send:match:trade');
|
|
}
|
|
self::$matchTradeData = []; //发送完清空,以避免重复发送相同的数据
|
|
}
|
|
|
|
|
|
public static function sendLeverHandle()
|
|
{
|
|
// echo date('Y-m-d H:i:s') . '定时器取价格' . PHP_EOL;
|
|
$now = microtime(true);
|
|
// $master_start = microtime(true);
|
|
// echo str_repeat('=', 80) . PHP_EOL;
|
|
// echo date('Y-m-d H:i:s') . '开始发送价格到杠杆交易系统' . PHP_EOL;
|
|
// echo '{' . PHP_EOL;
|
|
$market_kiline = self::$marketKlineData['1day'];
|
|
foreach ($market_kiline as $key => $value) {
|
|
$kline_data = $value['kline_data'];
|
|
$start = microtime(true);
|
|
// echo "\t" . date('Y-m-d H:i:s') . ' 发送' . $key . ',价格:' . $kline_data['close'] . PHP_EOL;
|
|
$params = [
|
|
'legal_id' => $kline_data['legal_id'],
|
|
'legal_name' => $kline_data['legal_name'],
|
|
'currency_id' => $kline_data['currency_id'],
|
|
'currency_name' => $kline_data['currency_name'],
|
|
'now_price' => $kline_data['close'],
|
|
'now' => $now
|
|
];
|
|
//价格大于0才进行任务推送
|
|
if (bc_comp($kline_data['close'], 0) > 0) {
|
|
LeverUpdate::dispatch($params)->onQueue('lever:update');
|
|
CoinTradeHandel::dispatch($params)->onQueue('coin:trade');
|
|
// LeverPushPrice::dispatch($params)->onQueue('lever:push:price');
|
|
}
|
|
$end = microtime(true);
|
|
// echo "\t" . date('Y-m-d H:i:s') . $key . '处理完成,耗时' .($end - $start) . '秒' . PHP_EOL;
|
|
}
|
|
// $master_end = microtime(true);
|
|
// echo '}' . PHP_EOL;
|
|
// echo date('Y-m-d H:i:s') . '杠杆交易系统处理完成,耗时' . ($master_end - $master_start) . '秒' . PHP_EOL;
|
|
// echo str_repeat('=', 80) . PHP_EOL;
|
|
}
|
|
|
|
public static function handleMicroTrade()
|
|
{
|
|
//self::$marketKlineData[$period][$key]['kline_data'] = $kline_data;
|
|
$market_data = self::$marketKlineData;
|
|
foreach ($market_data as $period => $data) {
|
|
foreach ($data as $key => $symbol) {
|
|
echo '秒合约时间:' . time() . ', Symbol:' . $key . '.' . $period . '数据' . PHP_EOL;
|
|
if ($period == '1min') {
|
|
//处理秒合约
|
|
// $match_id=$symbol['kline_data']['match_id'];
|
|
// $c_m=CurrencyMatch::find($match_id);
|
|
// if($c_m->open_microtrade == 1){
|
|
HandleMicroTrade::dispatch($symbol['kline_data'])->onQueue('micro_trade:handle');
|
|
HandleMicroTrade::dispatch($symbol['forward_data'])->onQueue('micro_trade:handle');//88888888
|
|
HandleMicroTrade::dispatch($symbol['foreign_data'])->onQueue('micro_trade:handle');
|
|
// }
|
|
|
|
} else {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public function writeMarketKline()
|
|
{
|
|
if ($this->worker_id < 8) {
|
|
$market_data = self::$marketKlineData;
|
|
foreach ($market_data as $period => $data) {
|
|
foreach ($data as $key => $symbol) {
|
|
echo '处理' . $key . '.' . $period . '数据' . PHP_EOL;
|
|
// file_put_contents('/www/wwwroot/crypto/public/d5.txt',$key . '-----' . $period ."---".json_encode($symbol).PHP_EOL,FILE_APPEND);
|
|
$result = MarketHour::getEsearchMarketById(
|
|
$symbol['kline_data']['currency_name'],
|
|
$symbol['market_data']['quote-currency'],
|
|
$period,
|
|
// $symbol['kline_data']['period'],
|
|
$symbol['market_data']['id']
|
|
);
|
|
if (isset($result['_source'])) {
|
|
$origin_data = $result['_source'];
|
|
// bc_comp($symbol['kline_data']['high'], $origin_data['high']) < 0
|
|
// && $symbol['kline_data']['high'] = $origin_data['high']; //新过来的价格如果不高于原最高价则不更新
|
|
// bc_comp($symbol['kline_data']['low'], $origin_data['low']) > 0
|
|
// && $symbol['kline_data']['low'] = $origin_data['low']; //新过来的价格如果不低于原最低价则不更新
|
|
}
|
|
|
|
$currencyInfo=Currency::where('name',$symbol['market_data']['base-currency'])->first();
|
|
if($currencyInfo){
|
|
$sisValue =$currencyInfo->oncontact;
|
|
if($sisValue>0){
|
|
$symbol['kline_data']['high'] = round($symbol['kline_data']['high'] + $sisValue,8);
|
|
$symbol['kline_data']['close'] = round($symbol['kline_data']['close'] + $sisValue,8);
|
|
$symbol['kline_data']['low'] = $symbol['kline_data']['low'] + $sisValue;
|
|
$symbol['market_data']['high'] = round($symbol['market_data']['high'] + $sisValue,8);
|
|
$symbol['market_data']['close'] = round($symbol['market_data']['close'] + $sisValue,8);
|
|
$symbol['market_data']['low'] = $symbol['market_data']['low'] + $sisValue;
|
|
if($period == '1day'){
|
|
$symbol['kline_data']['change'] = ($symbol['kline_data']['close'] - $symbol['kline_data']['open'])/$symbol['kline_data']['open']*100;
|
|
$symbol['kline_data']['change'] = round($symbol['kline_data']['change'],4);
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
SendMarket::dispatch($symbol['kline_data'])->onQueue('kline.all');
|
|
// EsearchMarket::dispatch($symbol['market_data'])->onQueue('esearch:market');//统一用一个队列
|
|
if ($period == '1min') {
|
|
// var_dump($symbol['kline_data']);
|
|
//推送一分钟行情
|
|
//SendMarket::dispatch($symbol['kline_data'])->onQueue('kline.1min');
|
|
//更新币种价格
|
|
UpdateCurrencyPrice::dispatch($symbol['kline_data'])->onQueue('update_currency_price');
|
|
} elseif ($period == '1day') {
|
|
//推送一天行情
|
|
$day_kline = $symbol['kline_data'];
|
|
$day_kline['type'] = 'kline';
|
|
// SendMarket::dispatch($day_kline)->onQueue('kline.all');
|
|
//SendMarket::dispatch($symbol['kline_data'])->onQueue('kline.1day');
|
|
//存入数据库
|
|
CurrencyQuotation::getInstance($symbol['kline_data']['legal_id'], $symbol['kline_data']['currency_id'])
|
|
->updateData([
|
|
'change' => $symbol['kline_data']['change'],
|
|
'now_price' => $symbol['kline_data']['close'],
|
|
'volume' => $symbol['kline_data']['volume'],
|
|
]);
|
|
} else {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
protected function calcIncreasePair($kline_data)
|
|
{
|
|
$open = $kline_data['open'];
|
|
$close = $kline_data['close'];;
|
|
$change_value = bc_sub($close, $open);
|
|
$change = bc_mul(bc_div($change_value, $open), 100, 2);
|
|
return $change;
|
|
}
|
|
|
|
//心跳响应
|
|
protected function onPong($con, $data)
|
|
{
|
|
//echo '收到心跳包,PING:' . $data->ping . PHP_EOL;
|
|
$send_data = [
|
|
'pong' => $data->ping,
|
|
];
|
|
$send_data = json_encode($send_data);
|
|
$con->send($send_data);
|
|
//echo '已进行心跳响应' . PHP_EOL;
|
|
}
|
|
|
|
public function ping($con)
|
|
{
|
|
$ping = time();
|
|
//echo '进程' . $this->worker_id . '发送ping服务器数据包,ping值:' . $ping . PHP_EOL;
|
|
$send_data = json_encode([
|
|
'ping' => $ping,
|
|
]);
|
|
$con->send($send_data);
|
|
// $this->pingTimer = Timer::add($this->server_time_out, function () use ($con) {
|
|
// $msg = '进程' . $this->worker_id . '服务器响应超时,连接关闭' . PHP_EOL;
|
|
// echo $msg;
|
|
// $this->close($msg);
|
|
// }, [], false);
|
|
}
|
|
|
|
protected function onPing($con, $data)
|
|
{
|
|
$this->pingTimer && Timer::del($this->pingTimer);
|
|
$this->pingTimer = null;
|
|
//echo '进程' . $this->worker_id . '服务器正常响应中,pong:' . $data->pong. PHP_EOL;
|
|
}
|
|
}
|