1289 lines
55 KiB
PHP
1289 lines
55 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\{CurrencyMatch,CurrencyQuotation,UsersWallet, MarketHour,Service\RedisService, UserChat, AccountLog, BindBox,BindBoxOrder,BindBoxQuotationLog,BindBoxSuccessOrder,BindBoxCollect,BindBoxMarginLog,BindBoxRaityHouse};
|
|
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 $bidding_nft;
|
|
|
|
protected $sendDepthTimer;
|
|
|
|
protected $depthTimer;
|
|
|
|
protected $handleTimer;
|
|
|
|
protected $microTradeHandleTimer;
|
|
|
|
protected $sendMatchTradeTimer;
|
|
|
|
protected $tzm = 0;
|
|
|
|
protected $tzmflag = 0;
|
|
|
|
protected $subscribed = [];
|
|
|
|
protected $bids = [];
|
|
|
|
protected $asks = [];
|
|
|
|
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', //成交的交易
|
|
],
|
|
];
|
|
|
|
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) {
|
|
// echo '处理期权订单' . PHP_EOL;
|
|
$this->bidding_nft = Timer::add(10, [$this, 'AuctionSettlementNFT'], [], true);//nft拍卖 结算 保证金退回
|
|
$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->bidding_nft && Timer::del($this->bidding_nft);
|
|
$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();
|
|
}
|
|
|
|
//nft拍卖 结算 保证金退回
|
|
public function AuctionSettlementNFT(){
|
|
$now = Carbon::now();
|
|
// 拍卖已经结束 没有人出价的拍卖品
|
|
$bind_boxs = BindBox::where('end_time','<=',$now)->where('pay_type',2)->get();
|
|
foreach ($bind_boxs as $bind_box){
|
|
$log = BindBoxQuotationLog::where('is_expired',0)->where('code',$bind_box->code)->first();
|
|
if(!$log){
|
|
$bind_box->status = 0;
|
|
$bind_box->resell_nft_status = 0;
|
|
$bind_box->save();
|
|
}
|
|
}
|
|
|
|
//拍卖结束 拍卖品有人出价
|
|
$bind_box_quotation_logs = DB::table('bind_box_quotation_log')
|
|
->leftjoin('bind_box_margin_log', 'bind_box_margin_log.id', '=', 'bind_box_quotation_log.margin_log_id')
|
|
->leftjoin('bind_box', 'bind_box.code', '=', 'bind_box_quotation_log.code')
|
|
->where('bind_box.end_time','<=',$now)
|
|
->where('bind_box.pay_type',2)
|
|
->where('bind_box.status',1)
|
|
->where('bind_box_quotation_log.is_expired',0)
|
|
->select('bind_box_quotation_log.*')->distinct()->get();
|
|
|
|
foreach ($bind_box_quotation_logs as $bind_box_quotation_log){
|
|
$quotation_log_id = $bind_box_quotation_log->id;
|
|
$status = $bind_box_quotation_log->status; //成交状态
|
|
if($status == 1){ //拍到了
|
|
echo '生成待支付订单_'.PHP_EOL;
|
|
$order = new BindBoxSuccessOrder();
|
|
$order->code = $bind_box_quotation_log->code;
|
|
$order->quotrtion_log_id = $bind_box_quotation_log->id;
|
|
$order->user_id = $bind_box_quotation_log->buyer_id;
|
|
$order->currency_id = $bind_box_quotation_log->currency_id;
|
|
$order->is_read = 0;
|
|
$order->is_pay = 0;
|
|
$order->overtime = time() + 86400;//过期时间24小时
|
|
$order->created = time();
|
|
$order->save();
|
|
|
|
Db::table('bind_box')->where('code',$bind_box_success_order->code)->update(['lock_order'=>1,'status'=>0]); //锁定未支付状态
|
|
Db::table('bind_box_quotation_log')->where('id',$quotation_log_id)->update(['is_expired'=>1]);
|
|
}else{//未拍到
|
|
$margin_log = BindBoxMarginLog::where('id',$bind_box_quotation_log->margin_log_id)->where('is_expired',0)->where('status',1)->first();//保证金交了未退
|
|
if($margin_log){//退保证金
|
|
$_currency = $margin_log->currency_id;
|
|
$number = $margin_log->number;
|
|
$users_wallet = UsersWallet::where('currency', $_currency)->where('user_id', $margin_log->user_id)->first();
|
|
change_wallet_balance($users_wallet, 2, $number, AccountLog::USER_RETURN_MARGIN_NFT, '退还竞拍保证金');
|
|
|
|
$margin_log->is_expired = 1;
|
|
$margin_log->status = 0;
|
|
$margin_log->save();
|
|
|
|
Db::table('bind_box_quotation_log')->where('id',$quotation_log_id)->update(['is_expired'=>1]);
|
|
}else{ //未拍到 但是推过保证金了
|
|
Db::table('bind_box_quotation_log')->where('id',$quotation_log_id)->update(['is_expired'=>1]);
|
|
}
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
//超时处理
|
|
$bind_box_success_orders = BindBoxSuccessOrder::where('overtime','<',time())->where('overtime','>',0)->where('is_expired',0)->where('is_pay',0)->get();
|
|
foreach ($bind_box_success_orders as $bind_box_success_order){
|
|
Db::table('bind_box')->where('code',$bind_box_success_order->code)->update(['status'=>0,'resell_nft_status'=>0,'lock_order'=>0]); //状态失效
|
|
$BindBoxQuotationLog = BindBoxQuotationLog::where('id',$bind_box_success_order->quotrtion_log_id)->first();
|
|
Db::table('bind_box_margin_log')->where('id',$BindBoxQuotationLog->margin_log_id)->update(['is_expired'=>1]);
|
|
$bind_box_success_order->is_expired = 1; //订单已失效
|
|
$bind_box_success_order->overtime = 0;
|
|
$bind_box_success_order->save();
|
|
}
|
|
}
|
|
|
|
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)
|
|
{
|
|
// $tzm += 1;
|
|
//加载配置文件 tasks_matches
|
|
$tasks_matches=getConfig("config/tasks_matches.php");
|
|
$sys_matches=$tasks_matches["tasks_matches"];//获取到“交易所”的交易对
|
|
foreach ($sys_matches as $k=>$v){
|
|
$self_match_key = $v['name'] . '.' . $v["legal_name"];
|
|
|
|
|
|
$to = time();
|
|
|
|
$t= DB::table('myquotation')->where('itime','<=', $to)->where('period','1min')->orderby('id','DESC')->first();
|
|
if(!empty($t)) {
|
|
// file_put_contents('/www/wwwroot/crypto/public/t1.txt',json_encode($t));
|
|
$f1 = "config/kline/".$t->itime."_kline_1min_".strtolower($v["name"]).".php";
|
|
$f2 = "config/kline/".$t->itime."_daymarket_".strtolower($v["name"]).".php";
|
|
|
|
|
|
if(!file_exists('/www/wwwroot/crypto/config/'.$f1)){
|
|
|
|
$f1 = "config/kline_1min_".strtolower($v["name"]).".php";
|
|
}
|
|
|
|
if(!file_exists('/www/wwwroot/crypto/config/'.$f2)){
|
|
|
|
$f2 = "config/daymarket_".strtolower($v["name"]).".php";
|
|
}
|
|
|
|
}else{
|
|
|
|
$f1 = "config/kline_1min_".strtolower($v["name"]).".php";
|
|
$f2 = "config/daymarket_".strtolower($v["name"]).".php";
|
|
}
|
|
|
|
// file_put_contents('/www/wwwroot/crypto/public/t1.txt',$f1.PHP_EOL,FILE_APPEND);
|
|
|
|
|
|
// $f1 = "config/kline_1min_".strtolower($v["name"]).".php";
|
|
// $f2 = "config/daymarket_".strtolower($v["name"]).".php";
|
|
|
|
$adt=getConfig($f1);
|
|
UserChat::sendChat($adt["libra_data"]);
|
|
// UserChat::sendChat($adt["libra_data"]);
|
|
$adt2=getConfig($f2);
|
|
UserChat::sendChat($adt2["libra_data"]);
|
|
|
|
$biname=$v["name"];
|
|
$biname2=strtolower($biname);
|
|
$rkey = "market.{$biname2}usdt.kline.1min";
|
|
$adt=$adt["libra_data"];
|
|
Redis::set($rkey, json_encode(['ch' => $rkey, 'ts' => $adt["time"], 'tick' => [
|
|
'id' => $adt["time"]/1000,
|
|
'open' => $adt["open"],
|
|
'close' => $adt["close"],
|
|
'low' => $adt["low"],
|
|
'high' => $adt["high"],
|
|
'vol' => $adt["volume"],
|
|
'amount' => $adt["volume"],
|
|
'count' => rand(10, 120)
|
|
]]));
|
|
$obj = json_decode(Redis::get($rkey));
|
|
if (!$obj) {
|
|
return; //必须 有这个obj才行的
|
|
}
|
|
$price=$adt["now_price"];
|
|
// if($tzm >= 5){
|
|
|
|
// }
|
|
// $bids = [];
|
|
// $asks = [];
|
|
// for ($i = 0; $i < 10; $i++) {
|
|
// $bids[] = [$price + rand(0, 9) * 0.00001, rand(50, 300)];
|
|
// }
|
|
// for ($i = 0; $i < 10; $i++) {
|
|
// $asks[] = [$price - rand(0, 9) * 0.00001, rand(50, 300)];
|
|
// }
|
|
// $depth_data22 = [
|
|
// 'type' => 'market_depth',
|
|
// 'symbol' => strtoupper($v["name"]) . '/'.$v["legal_name"],
|
|
// 'base-currency' =>$adt["currency_name"],
|
|
// 'quote-currency' => $v["legal_name"],
|
|
// 'currency_id' => $adt["currency_id"],
|
|
// 'currency_name' => $adt["currency_name"],
|
|
// 'legal_id' => $v["legal_id"],
|
|
// 'legal_name' => $v["legal_name"],
|
|
// 'bids' => $bids, //买入盘口
|
|
// 'asks' => $asks, //卖出盘口
|
|
// ];
|
|
|
|
// SendMarket::dispatch($depth_data22)->onQueue('market.depth');
|
|
|
|
//通知行情
|
|
$market_data_new =$adt;
|
|
$kline_data_new = $adt2["libra_data"];
|
|
|
|
$self_market_data = [
|
|
'id' => $adt["time"]/1000,
|
|
'period' => '1min',
|
|
'base-currency' => $adt["currency_name"],
|
|
'quote-currency' => $v["legal_name"],
|
|
'open' => sctonum($adt["open"]),
|
|
'close' => sctonum($adt["close"]),
|
|
'high' => sctonum($adt["high"]),
|
|
'low' => sctonum($adt["low"]),
|
|
'vol' => sctonum($adt["volume"]),
|
|
'amount' => sctonum($adt["close"])
|
|
];
|
|
|
|
$self_kline_data = [
|
|
'type' => 'kline',
|
|
'period' => "1min",
|
|
'match_id' => $v["id"],
|
|
'currency_id' => $kline_data_new["currency_id"],
|
|
'currency_name' => $kline_data_new["currency_name"],
|
|
'legal_id' => $kline_data_new["legal_id"],
|
|
'legal_name' => $kline_data_new["legal_name"],
|
|
'open' => sctonum($kline_data_new["open"]),
|
|
'close' => sctonum($kline_data_new["close"]),
|
|
'high' => sctonum($kline_data_new["high"]),
|
|
'low' => sctonum($kline_data_new["low"]),
|
|
'symbol' => $kline_data_new["symbol"],
|
|
'volume' => $kline_data_new["volume"],
|
|
'time' => $kline_data_new["time"]
|
|
];
|
|
|
|
/*
|
|
$params = [
|
|
'legal_id' => $kline_data_new["legal_id"],
|
|
'legal_name' => $kline_data_new["legal_name"],
|
|
'currency_id' => $kline_data_new["currency_id"],
|
|
'currency_name' => $kline_data_new["currency_name"],
|
|
'now_price' =>$kline_data_new["close"],
|
|
'now' => microtime(true)
|
|
];
|
|
//价格大于0才进行任务推送
|
|
if (bc_comp($kline_data_new["close"], 0) > 0) {
|
|
|
|
CoinTradeHandel::dispatch($params)->onQueue('coin:trade');
|
|
|
|
}
|
|
|
|
*/
|
|
|
|
self::$marketKlineData["1min"][$self_match_key] = ['market_data' => $self_market_data, 'kline_data' => $self_kline_data];
|
|
$self_data[] = [
|
|
'id' => time() . rand(1, 100000),
|
|
'amount' => rand(100, 39 * 56 - 184) / (16 * 56 + 104),
|
|
'direction' => "buy",
|
|
'price' => $price + rand(0, 9) * 0.00001, rand(50, 300),
|
|
'tradeId' => time() . rand(1, 3333),
|
|
'ts' => self::getMillisecond()
|
|
];
|
|
$self_data[] = [
|
|
'id' => time() . rand(1, 100000),
|
|
'amount' => rand(100, 39 * 56 - 184) / (16 * 56 + 104),
|
|
'direction' => "sell",
|
|
'price' => $price + rand(0, 9) * 0.00001, rand(50, 300),
|
|
'tradeId' => time() . rand(1, 3333),
|
|
'ts' => self::getMillisecond()
|
|
];
|
|
$self_trade_data = [
|
|
'type' => 'match_trade',
|
|
'symbol' => $kline_data_new["currency_name"] . '/' . $kline_data_new["legal_name"],
|
|
'base-currency' => $kline_data_new["currency_name"],
|
|
'quote-currency' => $kline_data_new["legal_name"],
|
|
'currency_id' => $kline_data_new["currency_id"],
|
|
'currency_name' => $kline_data_new["currency_name"],
|
|
'legal_id' => $kline_data_new["legal_id"],
|
|
'legal_name' => $kline_data_new["legal_name"],
|
|
'data' => $self_data
|
|
];
|
|
self::$matchTradeData[$kline_data_new["currency_name"] . '.' . $kline_data_new["legal_name"]] = $self_trade_data;
|
|
|
|
self::sendMatchTradeData();
|
|
|
|
CurrencyQuotation::getInstance($adt2["libra_data"]['legal_id'], $adt2["libra_data"]['currency_id'])
|
|
->updateData([
|
|
'change' => $adt2["libra_data"]['change'],
|
|
'now_price' => $adt2["libra_data"]['close'],
|
|
'volume' => $adt2["libra_data"]['volume'],
|
|
]);
|
|
}
|
|
|
|
// $tasks_matches=getConfig("config/tasks_matches.php");
|
|
// $sys_matches=$tasks_matches["tasks_matches"];//获取到“交易所”的交易对
|
|
// foreach ($sys_matches as $k=>$v){
|
|
// $adt=getConfig("config/kline_1min_".strtolower($v["name"]).".php");
|
|
// UserChat::sendChat($adt["libra_data"]);
|
|
|
|
// $biname=$v["name"];
|
|
// $biname2=strtolower($biname);
|
|
// $rkey = "market.{$biname2}usdt.kline.1min";
|
|
// $adt=$adt["libra_data"];
|
|
// Redis::set($rkey, json_encode(['ch' => $rkey, 'ts' => $adt["time"], 'tick' => [
|
|
// 'id' => $adt["time"]/1000,
|
|
// 'open' => $adt["open"],
|
|
// 'close' => $adt["close"],
|
|
// 'low' => $adt["low"],
|
|
// 'high' => $adt["high"],
|
|
// 'vol' => $adt["volume"],
|
|
// 'amount' => $adt["volume"],
|
|
// 'count' => rand(10, 120)
|
|
// ]]));
|
|
|
|
// $obj = json_decode(Redis::get($rkey));
|
|
// if (!$obj) {
|
|
// return; //必须 有这个obj才行的
|
|
// }
|
|
|
|
// $price=$adt["now_price"];
|
|
// $bids = [];
|
|
// $asks = [];
|
|
|
|
// for ($i = 0; $i < 10; $i++) {
|
|
// $bids[] = [$price + rand(0, 9) * 0.00001, rand(50, 300)];
|
|
// }
|
|
// for ($i = 0; $i < 10; $i++) {
|
|
// $asks[] = [$price - rand(0, 9) * 0.00001, rand(50, 300)];
|
|
// }
|
|
|
|
|
|
// $depth_data22 = [
|
|
// 'type' => 'market_depth',
|
|
// 'symbol' => strtoupper($v["name"]) . '/'.$v["legal_name"],
|
|
// 'base-currency' =>$adt["currency_name"],
|
|
// 'quote-currency' => $v["legal_name"],
|
|
// 'currency_id' => $adt["currency_id"],
|
|
// 'currency_name' => $adt["currency_name"],
|
|
// 'legal_id' => $v["legal_id"],
|
|
// 'legal_name' => $v["legal_name"],
|
|
// 'bids' => $bids, //买入盘口
|
|
// 'asks' => $asks, //卖出盘口
|
|
// ];
|
|
// // // print_r($depth_data22);
|
|
// SendMarket::dispatch($depth_data22)->onQueue('market.depth');
|
|
|
|
|
|
|
|
// // UserChat::sendChat($adt["libra_data"]);
|
|
// $adt2=getConfig("config/daymarket_".strtolower($v["name"]).".php");
|
|
// UserChat::sendChat($adt2["libra_data"]);
|
|
// }
|
|
|
|
$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'];
|
|
$tick = $data->tick;
|
|
|
|
if (strtoupper($currency_match->currency_name) == 'BTC' && $currency_match->legal_name == 'USDT') {
|
|
$redis = RedisService::getInstance(4);
|
|
$self_match_ids = self::getSelfMatch();
|
|
foreach ($self_match_ids as $id) {
|
|
$currencyMatch = CurrencyMatch::find($id);
|
|
if (!$currencyMatch) {
|
|
continue 1;
|
|
}
|
|
$self_match_key = $currencyMatch->currency_name . '.' . $currencyMatch->legal_name;
|
|
$fluctuate_max = $currencyMatch->fluctuate_max;
|
|
$fluctuate_min = $currencyMatch->fluctuate_min;
|
|
$close_price = $redis->get('close' . $self_match_key);
|
|
if ($close_price) {
|
|
$wTwpAmQ = rand(1, 1000000);
|
|
if ($wTwpAmQ > 950000) {
|
|
$elfIDDJ = [-0.0004, -0.0003, -0.0002, -0.0001, 0.0001, 0.0002, 0.0003, 0.0004];
|
|
$UBSQiHJ = array_rand($elfIDDJ);
|
|
$elfIDDJ = $elfIDDJ[$UBSQiHJ];
|
|
$elfIDDJ = bc_add(1, $elfIDDJ, 4);
|
|
} else {
|
|
$elfIDDJ = 1;
|
|
}
|
|
$close_price = bc_mul($close_price, $elfIDDJ, 8);
|
|
if ($close_price > $fluctuate_max) {
|
|
$close_price = $fluctuate_max;
|
|
}
|
|
if ($close_price < $fluctuate_min) {
|
|
$close_price = $fluctuate_min;
|
|
}
|
|
} else {
|
|
$close_price = rand($fluctuate_min * 1000, $fluctuate_max * (56 * 51 - 1856)) / (56 * 51 - 1856);
|
|
}
|
|
$high_price = $close_price;
|
|
$low_price = $close_price;
|
|
$open_price = $redis->get($self_match_key . $period . $tick->id);
|
|
if ($open_price) {
|
|
} else {
|
|
$open_price = $close_price;
|
|
$redis->set($self_match_key . $period . $tick->id, $close_price);
|
|
}
|
|
$redis->set('current_price_' . $currencyMatch->currency_name . $currencyMatch->legal_name, $close_price);
|
|
$redis->set('close' . $self_match_key, $close_price);
|
|
$self_market_data = [
|
|
'id' => $tick->id,
|
|
'period' => $period,
|
|
'base-currency' => $currencyMatch->currency_name,
|
|
'quote-currency' => $currencyMatch->legal_name,
|
|
'open' => sctonum($open_price),
|
|
'close' => sctonum($close_price),
|
|
'high' => sctonum($high_price),
|
|
'low' => sctonum($low_price),
|
|
'vol' => sctonum($tick->vol) * $id / 10,
|
|
'amount' => sctonum($tick->amount) * $id / 10
|
|
];
|
|
$self_kline_data = [
|
|
'type' => 'kline',
|
|
'period' => $period,
|
|
'match_id' => $currencyMatch->id,
|
|
'currency_id' => $currencyMatch->currency_id,
|
|
'currency_name' => $currencyMatch->currency_name,
|
|
'legal_id' => $currencyMatch->legal_id,
|
|
'legal_name' => $currencyMatch->legal_name,
|
|
'open' => sctonum($open_price),
|
|
'close' => sctonum($close_price),
|
|
'high' => sctonum($high_price),
|
|
'low' => sctonum($low_price),
|
|
'symbol' => $currencyMatch->currency_name . '/' . $currencyMatch->legal_name,
|
|
'volume' => sctonum($tick->amount) * $id / 10,
|
|
'time' => $tick->id * 1000
|
|
];
|
|
self::$marketKlineData[$period][$self_match_key] = ['market_data' => $self_market_data, 'kline_data' => $self_kline_data];
|
|
if ($period == '1day') {
|
|
$self_change = $this->calcIncreasePair($self_kline_data);
|
|
bc_comp($self_change, 0) > 0 && ($self_change = '+' . $self_change);
|
|
$self_daymarket_data = ['type' => 'daymarket', 'change' => $self_change, 'now_price' => $self_market_data['close'], 'api_form' => 'huobi_websocket'];
|
|
$self_kline_data = array_merge($self_kline_data, $self_daymarket_data);
|
|
self::$marketKlineData[$period][$self_match_key]['kline_data'] = $self_kline_data;
|
|
}
|
|
}
|
|
}
|
|
if ($currency_match->market_from == 2) {
|
|
// $hot = DB::table('currency_quotation')
|
|
// ->join('currency', 'currency_quotation.currency_id', '=', 'currency.id')
|
|
// ->where('currency.is_display', 1)
|
|
// ->where('currency.is_legal', 1)
|
|
// ->orderBy("currency_quotation.volume","desc")
|
|
// ->limit(6)
|
|
// ->get();
|
|
|
|
// $increase = DB::table('currency_quotation')
|
|
// ->join('currency', 'currency_quotation.currency_id', '=', 'currency.id')
|
|
// ->where('currency.is_display', 1)
|
|
// ->where('currency.is_legal', 1)
|
|
// ->orderBy("currency_quotation.change","desc")
|
|
// ->limit(6)
|
|
// ->get();
|
|
$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)//,
|
|
// 'hot' => $hot,
|
|
// 'increase' => $increase
|
|
];
|
|
|
|
$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//,
|
|
// 'hot' => $hot,
|
|
// 'increase' => $increase
|
|
];
|
|
//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,
|
|
];
|
|
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);
|
|
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);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function getMillisecond()
|
|
{
|
|
list($pSppfwv, $hmvXQMv) = explode(' ', microtime());
|
|
return (float)sprintf('%.0f', (floatval($pSppfwv) + floatval($hmvXQMv)) * (0 - 1352 + 42 * 56));
|
|
}
|
|
|
|
public static function getSelfMatch()
|
|
{
|
|
$redis = RedisService::getInstance(4);
|
|
$result = $redis->get('self_match');
|
|
if (!$result) {
|
|
$match_ids = CurrencyMatch::where('market_from', 0)->where('fluctuate_min', '>', 0)->where('fluctuate_max', '>', 0)->pluck('id')->toArray();
|
|
$redis->set('self_match', implode(',', $match_ids));
|
|
$redis->expire('self_match', 10);
|
|
return $match_ids;
|
|
} else {
|
|
return explode(',', $result);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 撮合交易全站交易数据回调
|
|
* @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;
|
|
if ($symbol_key == 'BTC.USDT') {
|
|
$self_match_ids = self::getSelfMatch();
|
|
foreach ($self_match_ids as $id) {
|
|
$currencyMatch = CurrencyMatch::query()->find($id);
|
|
if (!$currencyMatch) {
|
|
continue 1;
|
|
}
|
|
$now_price = CurrencyQuotation::where('match_id', $id)->value('now_price');
|
|
$rate_arr = [-0.004, -0.003, -0.002, -0.001, 0.001, 0.002, 0.003, 0.004];
|
|
$price = bc_add($now_price, $rate_arr[array_rand($rate_arr)]);
|
|
$self_data = [];
|
|
$self_data[] = [
|
|
'id' => time() . rand(1, 100000),
|
|
'amount' => rand(100, 39 * 56 - 184) / (16 * 56 + 104),
|
|
'direction' => $data[0]->direction,
|
|
'price' => $price,
|
|
'tradeId' => time() . rand(1, 3333),
|
|
'ts' => self::getMillisecond()
|
|
];
|
|
$self_trade_data = [
|
|
'type' => 'match_trade',
|
|
'symbol' => $currencyMatch->currency_name . '/' . $currencyMatch->legal_name,
|
|
'base-currency' => $currencyMatch->currency_name,
|
|
'quote-currency' => $currencyMatch->legal_name,
|
|
'currency_id' => $currencyMatch->currency_id,
|
|
'currency_name' => $currencyMatch->currency_name,
|
|
'legal_id' => $currencyMatch->legal_id,
|
|
'legal_name' => $currencyMatch->legal_name,
|
|
'data' => $self_data
|
|
];
|
|
self::$matchTradeData[$currencyMatch->currency_name . '.' . $currencyMatch->legal_name] = $self_trade_data;
|
|
}
|
|
}
|
|
if ($currency_match->market_from == 2) {
|
|
$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);
|
|
|
|
if ($currency_match->currency_name == 'BTC' && $currency_match->legal_name == 'USDT') {
|
|
$tick_bids = array_slice($data->tick->bids, 0, 20);
|
|
$tick_asks = array_slice($data->tick->asks, 0, 20);
|
|
$self_match_ids = self::getSelfMatch();
|
|
$redis = RedisService::getInstance(4);
|
|
foreach ($self_match_ids as $id) {
|
|
$currencyMatch = CurrencyMatch::query()->find($id);
|
|
if (!$currencyMatch) {
|
|
continue 1;
|
|
}
|
|
$current_price = $redis->get('current_price_' . $currencyMatch->currency_name . $currencyMatch->legal_name);
|
|
$current_price_new = bc_add($current_price, 0.0001 * count($tick_bids), 4);
|
|
$asks = [];
|
|
foreach ($tick_bids as $bid) {
|
|
$asks[] = [$current_price_new, bc_mul($bid[1], 3, 4)];
|
|
$current_price_new = bc_sub($current_price_new, 0.0001, 4);
|
|
}
|
|
$current_price_rate = bc_sub($current_price, 0.0001, 4);
|
|
$bids = [];
|
|
foreach ($tick_asks as $ask) {
|
|
$bids[] = [$current_price_rate, bc_mul($ask[1], 3, 4)];
|
|
$current_price_rate = bc_sub($current_price_rate, 0.0001, 4);
|
|
}
|
|
$symbol = $currencyMatch->currency_name . '/' . $currencyMatch->legal_name;
|
|
$self_depth_data = [
|
|
'type' => 'market_depth',
|
|
'symbol' => $symbol,
|
|
'base-currency' => $currencyMatch->currency_name,
|
|
'quote-currency' => $currencyMatch->legal_name,
|
|
'currency_id' => $currencyMatch->currency_id,
|
|
'currency_name' => $currencyMatch->currency_name,
|
|
'legal_id' => $currencyMatch->legal_id,
|
|
'legal_name' => $currencyMatch->legal_name,
|
|
'bids' => $bids,
|
|
'asks' => $asks
|
|
];
|
|
self::$marketDepthData[$symbol] = $self_depth_data;
|
|
}
|
|
}
|
|
if ($currency_match->market_from == 2) {
|
|
$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, 20), //买入盘口
|
|
'asks' => array_slice($data->tick->asks, 0, 20), //卖出盘口
|
|
];
|
|
$symbol_key = $currency_match->currency_name . '.' . $currency_match->legal_name;
|
|
self::$marketDepthData[$symbol_key] = $depth_data;
|
|
|
|
if($currency_match->currency_name == "BTC"){
|
|
$tasks_matches=getConfig("config/tasks_matches.php");
|
|
|
|
$to = time();
|
|
|
|
$t= DB::table('myquotation')->where('itime','<=', $to)->where('period','1min')->orderby('id','DESC')->first();
|
|
|
|
|
|
|
|
|
|
|
|
$sys_matches=$tasks_matches["tasks_matches"];//获取到“交易所”的交易对
|
|
foreach ($sys_matches as $k=>$v){
|
|
if(!empty($t))
|
|
{
|
|
$f1 = "config/kline/".$t->itime."_kline_1min_".strtolower($v["name"]).".php";
|
|
// $adt=getConfig("config/kline_1min_".strtolower($v["name"]).".php");
|
|
|
|
if(!file_exists('/www/wwwroot/crypto/config/'.$f1)){
|
|
|
|
$f1 = "config/kline_1min_".strtolower($v["name"]).".php";
|
|
}
|
|
}else{
|
|
|
|
$f1 = "config/kline_1min_".strtolower($v["name"]).".php";
|
|
}
|
|
|
|
|
|
$adt=getConfig($f1);
|
|
$adt = $adt["libra_data"];
|
|
$price=$adt["now_price"];
|
|
|
|
for ($i = 0; $i < 20; $i++) {
|
|
$bids[] = [$price + rand(0, 2) * 0.00001, rand(50, 300)];
|
|
}
|
|
for ($i = 0; $i < 20; $i++) {
|
|
$asks[] = [$price - rand(0, 2) * 0.00001, rand(50, 300)];
|
|
}
|
|
$depth_data22 = [
|
|
'type' => 'market_depth',
|
|
'symbol' => strtoupper($v["name"]) . '/'.$v["legal_name"],
|
|
'base-currency' =>$adt["currency_name"],
|
|
'quote-currency' => $v["legal_name"],
|
|
'currency_id' => $adt["currency_id"],
|
|
'currency_name' => $adt["currency_name"],
|
|
'legal_id' => $v["legal_id"],
|
|
'legal_name' => $v["legal_name"],
|
|
'bids' => $bids, //买入盘口
|
|
'asks' => $asks, //卖出盘口
|
|
];
|
|
|
|
SendMarket::dispatch($depth_data22)->onQueue('market.depth');
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
/**
|
|
* 发送盘口数据
|
|
*
|
|
* @return void
|
|
*/
|
|
public function sendDepthData()
|
|
{
|
|
$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);
|
|
|
|
// file_put_contents("onMessage.txt", json_encode($data) . PHP_EOL, FILE_APPEND);
|
|
|
|
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');
|
|
// }
|
|
|
|
} 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;
|
|
$result = MarketHour::getEsearchMarketById(
|
|
$symbol['market_data']['base-currency'],
|
|
$symbol['market_data']['quote-currency'],
|
|
$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']; //新过来的价格如果不低于原最低价则不更新
|
|
}
|
|
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;
|
|
}
|
|
}
|