<?php
namespace app\api\controller;
use app\common\controller\Api;
use app\common\model\TradeConfig;
use think\worker\Server;
use Workerman\Worker;
use Workerman\Lib\Timer;
use Workerman\Connection\AsyncTcpConnection;
use \Workerman\Autoloader;
use GatewayWorker\Gateway;
use think\Log;
use think\Db;
use think\Exception;
use fast\Random;
use app\common\library\Token;


//use think\console\Input;
//use think\console\Output;
//use think\console\Command;

//心跳间隔5秒
define('HEARTBEAT_TIME', 10);

/**
 *交易对K线图数据生成-火币
 * 用于实时接收火币K线数据
 */
class NewTradeKlines extends Api
{
    protected $noNeedLogin = ['*'];
    protected $noNeedRight = ['*'];

    protected $flag = true;//是否正式环境

    protected $huobi_host = '';

    protected $server_host = 'ws://api.huobi.pro/ws';

    protected $host = 'ws://api.huobi.pro/ws';

    protected $local_host = 'Websocket://0.0.0.0:17878';// 代理监听本地9999端口

    private $time_list = [
        '1min'=>60, //1分钟
        '5min'=>300,//5分钟
        '15min'=>900,//15分钟
        '30min'=>1800,//30分钟
        '60min'=>3600,//1小时
        '1day'=>86400,//1天
        '1week'=>604800,//1周
        '1mon'=>2592000, //1月
        //'1year'=>31536000, //1年
    ];

    private $time_lists = ['1min','5min','15min','30min','1day','1week','1mon'];

    private $symbol_list = ['market.btcusdt.trade.detail',];

    private $all_cons = [];

    private $all_symbols = ['btcusdt','ethusdt','ltcusdt','bchusdt','eosusdt'];

    //private $all_dic = [];

    private $testip = array("192.168.10.234", "192.168.230.1");

    private $huobi_id = 0;//连接火币服务器的连接id，防止心跳把火币连接关闭

    private $reconnect_num = 0;//与火币服务器的重连次数，超过一定次数重启Worker，目前是10次，windows下无法重启

    private $async_message_time = 0;//与火币服务器的消息交互时间，超过一定时间没有消息往来重启Worker，目前是300s，windows下无法重启

    public function index()
    {
        // 创建一个Worker监听2345端口，使用http协议通讯
        $context = array(
            // 更多ssl选项请参考手册 http://php.net/manual/zh/context.ssl.php
            'ssl' => array(
                // 请使用绝对路径
                'local_cert'                 => '/www/wwwroot/ws.btcex.tw/server/server.pem', // 也可以是crt文件
                'local_pk'                   => '/www/wwwroot/ws.btcex.tw/server/server.key',
                'verify_peer'                => false,
                'allow_self_signed' => true, //如果是自签名证书需要开启此选项
            )
        );
         
        // Worker::$stdoutFile = '/www/wwwroot/196api/public/uploads/logs/ntkline.log';
        $this->worker = new Worker("websocket://0.0.0.0:17878",$context); //
//        $this->worker = new Worker("websocket://0.0.0.0:17878"); //
        $this->worker->transport = 'ssl';
        $info = "启动Worker-start:".date('Y-m-d H:i:s');
        echo "\r\n ".$info;
//        $this->saveLog("huobi", $info);
        $this->ctrl = [];
        $this->userctrldy = [];
        $this->userctrl = [];
        $this->historykline = [];
        // 启动1个进程对外提供服务
        $this->worker->count = 1;
        $this->worker->name = 'huobikline';
        $this->coins = Db::name('app_rate')
            ->where('is_hytrade','1')
            ->select();
        $this->worker->onWorkerStart = function($worker)
        {
            $this->onWorkerStart($worker);
        };

        $this->huobiflag = false;
        $this->userdy = [];
        $this->usersd = [];
        $this->ztc_history = [];
        $this->ztc_depth = [];
        $this->ztc_tradenow = [];
        $this->redis = getRedis();
        // 接收到浏览器发送的数据时回复hello world给浏览器
        $this->worker->onMessage = function($connection, $data)
        {
            $this->onWorkerMessage($connection, $data);
        };

        Worker::runAll();
    }

    function onWorkerStart($worker)
    {
        $info = "启动Worker-start success:".date('Y-m-d H:i:s');
        echo "\r\n ".$info;
//        $this->saveLog("huobi", $info);
        // 进程启动后设置一个每秒运行一次的定时器
        Timer::add(1, function()use($worker){
            $time_now = time();
            if(count($worker->connections) > 0) {
//                $this->saveLog("all", '心跳计时器,count:' . count($worker->connections));
            }
            foreach($worker->connections as $connection) {
                if ($connection->id == $this->huobi_id) {
//                    $this->saveLog("all", '心跳计时器,huobi_id:'.$this->huobi_id);
                    continue;
                }
                // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
                if (empty($connection->lastMessageTime)) {
                    $connection->lastMessageTime = $time_now;
                    continue;
                }
                // 上次通讯时间间隔大于心跳间隔*2，则认为客户端已经下线，关闭连接
                if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME * 2) {
//                    $this->saveLog("all", '心跳计时器,心跳超时,cid:'.$connection->id.',now:'.date('Y-m-d H:i:s', $time_now).',lastMessageTime:'.date('Y-m-d H:i:s', $connection->lastMessageTime));
                    $connection->close();
                    //unset($this->all_cons[$connection->id]);
                }
            }
            
            //循环发送APT深度及K线
            $user_arr = [];
            if(isset($this->userdy['market.aptusdt.kline.1min'])){
                foreach ($this->userdy['market.aptusdt.kline.1min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.aptusdt.kline.1min';
                    $this->apt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.aptusdt.kline.5min'])){
                foreach ($this->userdy['market.aptusdt.kline.5min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.aptusdt.kline.5min';
                    $this->apt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.aptusdt.kline.15min'])){
                foreach ($this->userdy['market.aptusdt.kline.15min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.aptusdt.kline.15min';
                    $this->apt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.aptusdt.kline.30min'])){
                foreach ($this->userdy['market.aptusdt.kline.30min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.aptusdt.kline.30min';
                    $this->apt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.aptusdt.kline.60min'])){
                foreach ($this->userdy['market.aptusdt.kline.60min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.aptusdt.kline.60min';
                    $this->apt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.aptusdt.kline.1day'])){
                foreach ($this->userdy['market.aptusdt.kline.1day'] as $key=>$value){
                    $data_ls['sub'] = 'market.aptusdt.kline.1day';
                    if(in_array($value, $user_arr)){
                        $this->apt_send($data_ls,$worker->connections,$value,false);
                    }else{
                        $this->apt_send($data_ls,$worker->connections,$value);
                    }
                }
            }
            //EVT的
            if(isset($this->userdy['market.ttdusdt.kline.1min'])){
                foreach ($this->userdy['market.ttdusdt.kline.1min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.ttdusdt.kline.1min';
                    $this->evt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.ttdusdt.kline.5min'])){
                foreach ($this->userdy['market.ttdusdt.kline.5min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.ttdusdt.kline.5min';
                    $this->evt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.ttdusdt.kline.15min'])){
                foreach ($this->userdy['market.ttdusdt.kline.15min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.ttdusdt.kline.15min';
                    $this->evt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.ttdusdt.kline.30min'])){
                foreach ($this->userdy['market.ttdusdt.kline.30min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.ttdusdt.kline.30min';
                    $this->evt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.ttdusdt.kline.60min'])){
                foreach ($this->userdy['market.ttdusdt.kline.60min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.ttdusdt.kline.60min';
                    $this->evt_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.ttdusdt.kline.1day'])){
                foreach ($this->userdy['market.ttdusdt.kline.1day'] as $key=>$value){
                    $data_ls['sub'] = 'market.ttdusdt.kline.1day';
                    if(in_array($value, $user_arr)){
                        $this->evt_send($data_ls,$worker->connections,$value,false);
                    }else{
                        $this->evt_send($data_ls,$worker->connections,$value);
                    }
                }
            }
            
            //IFT的
            if(isset($this->userdy['market.iftusdt.kline.1min'])){
                foreach ($this->userdy['market.iftusdt.kline.1min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.iftusdt.kline.1min';
                    $this->ift_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.iftusdt.kline.5min'])){
                foreach ($this->userdy['market.iftusdt.kline.5min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.iftusdt.kline.5min';
                    $this->ift_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.iftusdt.kline.15min'])){
                foreach ($this->userdy['market.iftusdt.kline.15min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.iftusdt.kline.15min';
                    $this->ift_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.iftusdt.kline.30min'])){
                foreach ($this->userdy['market.iftusdt.kline.30min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.iftusdt.kline.30min';
                    $this->ift_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.iftusdt.kline.60min'])){
                foreach ($this->userdy['market.iftusdt.kline.60min'] as $key=>$value){
                    $user_arr[] = $value;
                    $data_ls['sub'] = 'market.iftusdt.kline.60min';
                    $this->ift_send($data_ls,$worker->connections,$value);
                }
            }
            if(isset($this->userdy['market.iftusdt.kline.1day'])){
                foreach ($this->userdy['market.iftusdt.kline.1day'] as $key=>$value){
                    $data_ls['sub'] = 'market.iftusdt.kline.1day';
                    if(in_array($value, $user_arr)){
                        $this->ift_send($data_ls,$worker->connections,$value,false);
                    }else{
                        $this->ift_send($data_ls,$worker->connections,$value);
                    }
                }
            }


        });
        
        Timer::add(5, function()use($worker){
             $ctrls = Db::name('user')
                 ->field('id,tradectrl_json')
                 ->select();
             $this->userctrl = [];
             if($ctrls) {
                 foreach ($ctrls as $key => $value) {
                     $this->userctrl[$value['id']] = $value['tradectrl_json'];
                 }
             }
        });

        // 异步建立一个到火币服务器的连接
        $con = new AsyncTcpConnection($this->host);
        $this->cons = $con;
        if ($this->flag) {//正式环境
            $con->transport = 'ssl';
        }

        // 当服务器连接发来数据时，转发给对应客户端的连接
        $con->onMessage = function($con, $message) use($worker)
        {
            $this->onAsyncMessage($con, $message, $worker);
        };

        $con->onError = function($con, $err_code, $err_msg)
        {
//                var_dump(6);
            echo "$err_code, $err_msg";
            $info = "Async onError err_code:{$err_code},err_msg:{$err_msg}";
            echo "\r\n ".$info;
//            $this->saveLog("huobi", $info);
        };

        $con->onClose = function($con)
        {
//            $this->saveLog("huobi", '火币连接断开，正在重连');
            // 如果连接断开，则在1秒后重连

            $this->onWorkerStart($this->worker);
            $con->reConnect(1);
        };

        $con->connect();

        //var_dump(1);
    }

    function onWorkerMessage($connection, $data)
    {
        // 给connection临时设置一个lastMessageTime属性，用来记录上次收到消息的时间
        $connection->lastMessageTime = time();
        $data = json_decode($data, true);
        $connection->lastMessageTime = time();
        if(isset($data['pong'])) {//客户端返回心跳pong
            $connection->send(json_encode(array('pong success')));
        }else if(isset($data['subs']) && $data['subs'] == 'history' && isset($data['symbol']) && strpos($data['symbol'], 'ztcusdt') === false) {
            $from = time() - $this->time_list[$data['period']] * $data['size'];
            $to = time();
            $datas = [
                'req' => "market.".$data['symbol'].".kline.".$data['period'],
                'id' => 'id'.time(),
                'from' => $from,
                'to' => $to
            ];
            $this->historykline[$datas['id']] = $connection->id;
            $this->cons->send(json_encode($datas));
            $info = "\r\n cid ".$connection->id."订阅K线历史".json_encode($data);//."--".json_encode($result);
            echo $info;
//            $this->saveLog("all", $info);
        }else if(((isset($data['subs']) && $data['subs'] == 'tradenow') && isset($data['sub']) && strpos($data['sub'], 'ztcusdt') === false) || ($data['subs'] == 'tradenow' && isset($data['unsub']))){
            $userdy = $this->userdy;
            if(isset($data['sub'])) {
                $this->userdy[$data['sub']][] = $connection->id;
                if(isset($data['user_id'])) {
                    $this->userctrldy[$data['user_id']] = $connection->id;
                }
            }

            if(isset($data['unsub']) && isset($this->userdy[$data['unsub']]))
            {
                $keys = array_search($connection->id, $this->userdy[$data['unsub']]);
                if($keys>=0) unset($this->userdy[$data['unsub']][$keys]);
                $info = "\r\n cid " . $connection->id . "取消k线数据" . json_encode($data);
                
                //删除特定的定时器
                // if(strpos($data['unsub'],'aptusdt') !== false){
                //     Timer::del($this->timer_user[$connection->id]);
                // }
                // $info .= "删除定时器成功";
                echo $info;
            }else if(isset($data['sub'])){
//            var_dump($data);
                $info = "\r\n cid " . $connection->id . "订阅k线数据" . json_encode($data);
                echo $info;
                
                //创建一个特定定时器
                if(strpos($data['sub'],'aptusdt') !== false){
                    
                }else{
                    $sub_ch = explode(".", $data['sub']);
                    $release = Db::name("app_curr_release")->where("symbol",$sub_ch[1])->where("status","in","3")->find();
                    if($release){
                        $sub_ch[1] = $release['hb_symbol'];
                        $sub_str = implode(".", $sub_ch);
                        if(!isset($userdy[$sub_str])) {
                            $data = [
                                'sub' => $sub_str,
                                "id" => "id" . time(),
                            ];
                            $this->cons->send(json_encode($data));
                        }
                    }else{
                        if(!isset($userdy[$data['sub']])) {
                            $data = [
                                'sub' => $data['sub'],
                                "id" => "id" . time(),
                            ];
                            $this->cons->send(json_encode($data));
                        }
                    }
                }
                
            }

        }else if(($data['subs'] == 'depth' && isset($data['sub']) && strpos($data['sub'], 'ztcusdt') === false) || ($data['subs'] == 'depth' && isset($data['unsub']))){
            $usersd = $this->usersd;
            if(isset($data['sub'])) {
                $this->usersd[$data['sub']][] = $connection->id;
            }
            if(isset($data['unsub']) && isset($this->usersd[$data['unsub']]))
            {
                $keys = array_search($connection->id, $this->usersd[$data['unsub']]);
                if($keys>=0) unset($this->usersd[$data['unsub']][$keys]);
            }else if(isset($data['sub'])){
//            var_dump($data);
                $info = "\r\n cid " . $connection->id . "订阅k线深度" . json_encode($data);
                echo $info;
                
                $sub_ch = explode(".", $data['sub']);
                $release = Db::name("app_curr_release")->where("symbol",$sub_ch[1])->where("status","in","3")->find();
                if($release){
                    $sub_ch[1] = $release['hb_symbol'];
                    $sub_str = implode(".", $sub_ch);
                    if(!isset($usersd[$sub_str])) {
                        $data = [
                            'sub' => $sub_str,
                            "id" => "id" . time(),
                        ];
                        $this->cons->send(json_encode($data));
                    }
                }else{
                    if(!isset($usersd[$data['sub']])) {
                        $data = [
                            'sub' => $data['sub'],
                            "id" => "id" . time(),
                        ];
                        $this->cons->send(json_encode($data));
                    }
                }
            }

        }else if($data['subs'] == 'ztc-history') {

        }else if($data['subs'] == 'ztc-depth') {

        }else if($data['subs'] == 'ztc-tradenow') {

        }

    }
    
    /**
     * APT深度数据和实时k线 
     */
    function apt_send($data,$connection,$id,$is_sd=true){
        $sub_arr = explode(".", $data['sub']);
        $kline = $this->redis->get("apt_".$sub_arr[3]);
        $apt_kline = json_decode($kline,true);
//        var_dump($apt_kline);
        $kline_data = array(
            "ch" => $data['sub'],//"market.aptusdt.kline.1min",
            "subs" => "tradenow",
            "tick" => $apt_kline,
        );
        // echo "\r\n发送ID：".$id." 成功";
        $connection[$id]->send(json_encode($kline_data));
        if($is_sd){
            $bids = json_decode($this->redis->get("buy_shendu"),true);
            $asks = json_decode($this->redis->get("sell_shendu"),true);
            $depth_data = array(
                "ch" => "market.aptusdt.depth.step0",
                "subs" => "depth",
                "tick" => [
                    "asks" => $asks,
                    "bids" => $bids,
                ],
            );
            $connection[$id]->send(json_encode($depth_data));
        }
    }
    
    /**
     * EVT深度数据和实时k线 
     */
    function evt_send($data,$connection,$id,$is_sd=true){
        $sub_arr = explode(".", $data['sub']);
        $kline = $this->redis->get("evt_".$sub_arr[3]);
        $evt_kline = json_decode($kline,true);
//        var_dump($evt_kline);
        $kline_data = array(
            "ch" => $data['sub'],//"market.evtusdt.kline.1min",
            "subs" => "tradenow",
            "tick" => $evt_kline,
        );
        // echo "\r\n发送ID：".$id." 成功";
        $connection[$id]->send(json_encode($kline_data));
        if($is_sd){
            $bids = json_decode($this->redis->get("buy_shendu_evt"),true);
            $asks = json_decode($this->redis->get("sell_shendu_evt"),true);
            $depth_data = array(
                "ch" => "market.ttdusdt.depth.step0",
                "subs" => "depth",
                "tick" => [
                    "asks" => $asks,
                    "bids" => $bids,
                ],
            );
            $connection[$id]->send(json_encode($depth_data));
        }
    }
    
    /**
     * IFT深度数据和实时k线 
     */
    function ift_send($data,$connection,$id,$is_sd=true){
        $sub_arr = explode(".", $data['sub']);
        $kline = $this->redis->get("ift_".$sub_arr[3]);
        $ift_kline = json_decode($kline,true);
//        var_dump($ift_kline);
        $kline_data = array(
            "ch" => $data['sub'],//"market.iftusdt.kline.1min",
            "subs" => "tradenow",
            "tick" => $ift_kline,
        );
        // echo "\r\n发送ID：".$id." 成功";
        $connection[$id]->send(json_encode($kline_data));
        if($is_sd){
            $bids = json_decode($this->redis->get("buy_shendu_ift"),true);
            $asks = json_decode($this->redis->get("sell_shendu_ift"),true);
            $depth_data = array(
                "ch" => "market.iftusdt.depth.step0",
                "subs" => "depth",
                "tick" => [
                    "asks" => $asks,
                    "bids" => $bids,
                ],
            );
            $connection[$id]->send(json_encode($depth_data));
        }
    }

    /**
     *
     * @param type $url
     * @param type $type
     * @param type $arr
     * @return type
     */
    function http_curl($url, $type = 'get', $arr = '') {
        if($arr){
            $o = "";
            foreach ( $arr as $k => $v )
            {
                $o.= "$k=" . urlencode( $v ). "&" ;
            }
            $arr = substr($o,0,-1);
        }

        $ch = curl_init();

        $user_agent = "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/33.0.1750.146 Safari/537.36";
        curl_setopt($ch, CURLOPT_USERAGENT,$user_agent);
        curl_setopt($ch, CURLOPT_URL, $url); //设置访问的地址
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1); //获取的信息返回
//        curl_setopt($ch, CURLOPT_PROXY, "hk2.cable-modem.org"); //代理服务器地址
//        curl_setopt($ch, CURLOPT_PROXYPORT,"46543"); //代理服务器端口
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, false);
        curl_setopt($ch, CURLOPT_TIMEOUT, 20000);
        if ($type == 'post') {
            curl_setopt($ch, CURLOPT_POST, 1);
            curl_setopt($ch, CURLOPT_POSTFIELDS, $arr);
        }
        $output = curl_exec($ch);
        if (curl_error($ch)) {
            return curl_error($ch);
        }
        return $output;
    }

    function onAsyncConnect($con) {

        $this->async_message_time = time();
        $this->huobi_id = $con->id;

        foreach ($this->all_symbols as $key=>$value)
        {
            foreach ($this->time_lists as $k=>$val)
            {
                $data = [
                    'sub' => "market.".$value.".kline.".$val,
                    "id" => "id".time(),
                ];
                $this->userdy[$data['sub']] = [];
//                $this->saveLog("all", '异步连接火币，订阅:'.$data['sub'].'-'.json_encode($this->userdy));

                $con->send(json_encode($data));
            }
        }

    }

    function onAsyncMessage($con, $message, $worker)
    {
        $data = json_decode($message, true);
        if (!$data) {//说明采用了GZIP压缩
            $data = gzdecode($message);
//            $this->saveLog("huobi", $data);
            $data = json_decode($data, true);
        }
//        var_dump($data);  火币发送过来的数据
        if(isset($data['ping'])) {
            $this->async_message_time = time();
            $con->send(json_encode([
                "pong" => $data['ping']
            ]));
           foreach($worker->connections as $connection) {
               $connection->send(json_encode($data));
           }
        }else if (isset($data['ch'])) {
            $this->async_message_time = time();
            if(strpos($data['ch'],'kline') !== false) { //最新k线数据  ch格式：market.btcusdt.kline.1min
                $data['subs'] = 'tradenow';
//            echo "<pre>";
//            var_dump($this->userdy);
                $symbol = '';
                if($this->coins) {
                    foreach ($this->coins as $kk => $vv) {
                        if (strpos($data['ch'], $vv['symbol']) !== false) {
                            $symbol = $vv['symbol'];
                            $jd = $vv['jd'];
                            break;
                        }
                    }
                }
                $hour = date('H');
                $minute = date('i');
                if ($symbol) {
                    $coin = Db::name('app_rate')
                        ->where('symbol',$symbol)->find();
                    if($coin) {
                        $tradejson = json_decode($coin['tradectrl_json'],true);
                        $price = $num = 0;
                        if(!empty($tradejson)) {
                            foreach ($tradejson as $key => $value) {
                                $time1 = explode('-', $key);
                                $time2 = explode(':', $time1[0]);
                                $time3 = explode(':', $time1[1]);
                                $pricearr = explode('-', $value);
                                if ($hour >= $time2[0] && $hour <= $time3[0]) {
                                    if (($hour == $time3[0] && $minute > $time3[1]) || ($hour == $time2[0] && $minute < $time2[1])) continue;
                                    $price = (float)$this->randomFloat($pricearr[0], $pricearr[1], $jd);
                                    $num = $this->randomFloat(1, 10, 6);
                                    break;
                                }
                            }
                            if ($price > 0) {
                                $data['tick']['close'] = $price;
                            }
                        }
                    }
                }

                if($symbol && $this->ctrl && isset($this->ctrl[$symbol]) && isset($this->ctrl[$symbol][time()]))
                {
                    $data['tick']['close'] = (float)sprintf("%.2f",$this->ctrl[$symbol][time()]);
//                    $this->saveLog("all", '调控数据推送:'.json_encode($data));
                }

                //给真实订阅用户推送  最新一条数据
                if(isset($this->userdy[$data['ch']])){
                    foreach ($this->userdy[$data['ch']] as $key => $value) {
                        if (!isset($worker->connections[$value])) {
                            unset($this->userdy[$data['ch']][$key]);
                        } else {
                            if(in_array($value,$this->userctrldy)){
                                $datass = $data;
                                $userids = array_search($value, $this->userctrldy);
                                $pricearr = [];
                                if(isset($this->userctrl[$userids]))
                                {
                                    $tradejson = json_decode($this->userctrl[$userids],true);
                                    if($tradejson) {
                                        foreach ($tradejson as $kk => $vv) {
                                            if ($symbol && $symbol == $kk) {
                                                $jsonarr = explode('|', $vv);
                                                $time1 = explode('-', $jsonarr[0]);
                                                $time2 = explode(':', $time1[0]);
                                                $time3 = explode(':', $time1[1]);
                                                if ($hour >= $time2[0] && $hour <= $time3[0]) {
                                                    if (($hour == $time3[0] && $minute > $time3[1]) || ($hour == $time2[0] && $minute < $time2[1])) continue;
                                                    $pricearr = explode('-', $jsonarr[1]);
                                                    break;
                                                }
                                            }
                                        }
                                    }
                                }
                                if($pricearr){
                                    $price = (float)$this->randomFloat($pricearr[0], $pricearr[1], $jd);
                                    if ($price > 0) {
                                        $datass['tick']['close'] = $price;
                                    }
                                }
                                $worker->connections[$value]->send(json_encode($datass));
                            }else{
                                $worker->connections[$value]->send(json_encode($data));
                            }
                        }
                    }
                }
                
                //给模拟订阅用户推送
                $ch_kline = explode(".", $data['ch']);
                $release = Db::name("app_curr_release")->where("hb_symbol",$ch_kline[1])->where("status","in","3")->select();
                if($release){
                    foreach ($release as $kkk => $vvvv) {
                        $ch_kline[1] = $vvvv['symbol'];
                        $ch_str = implode(".", $ch_kline);
                        if($vvvv['is_tk'] == 1){
                            //调控
                        }
                        if(isset($this->userdy[$ch_str])){
                            $data['ch'] = $ch_str;
                            foreach ($this->userdy[$ch_str] as $key => $value) {
                                if (!isset($worker->connections[$value])) {
                                    unset($this->userdy[$ch_str][$key]);
                                } else {
                                    $worker->connections[$value]->send(json_encode($data));
                                }
                            }
                        }
                    }
                }
            }else if(strpos($data['ch'],'depth') !== false) //最新深度数据  ch格式：market.btcusdt.depth.step0
            {
                $data['subs'] = 'depth';
                $symbol = '';
                foreach ($this->coins as $kk => $vv)
                {
                    if (strpos($data['ch'], $vv['symbol']) !== false){
                        $symbol = $vv['symbol'];
                        $jd = $vv['jd'];
                        break;
                    }
                }
                $hour = date('H');
                $minute = date('i');
                if ($symbol) {
                    $coin = Db::name('app_rate')
                        ->where('symbol',$symbol)->find();
                    if($coin) {
                        $tradejson = json_decode($coin['tradectrl_json'],true);
                        $price = $num = 0;
                        $pricearrs = [];
                        if(!empty($tradejson)) {
                            foreach ($tradejson as $key => $value) {
                                $time1 = explode('-', $key);
                                $time2 = explode(':', $time1[0]);
                                $time3 = explode(':', $time1[1]);
                                $pricearr = explode('-', $value);
                                if ($hour >= $time2[0] && $hour <= $time3[0]) {
                                    if (($hour == $time3[0] && $minute > $time3[1]) || ($hour == $time2[0] && $minute < $time2[1])) continue;
                                    $price = (float)$this->randomFloat($pricearr[0], $pricearr[1], $jd);
                                    $num = $this->randomFloat(1, 10, 6);
                                    $pricearrs = $pricearr;
                                    break;
                                }
                            }
                            if (!empty($pricearrs)) {
                                //买一价格
                                $bidsprice = $pricearrs[0];
                                $bidsprices = $bidsprice;
                                //卖一价格
                                $asksprice = $pricearrs[1];
                                $asksprices = $asksprice;
                                foreach ($data['tick']['bids'] as $key=>$value)
                                {
                                    $value[0] = $bidsprices - (float)$this->randomFloat(1, 5, $jd);
                                    $data['tick']['bids'][$key] = $value;
                                    $bidsprices = $value[0];
                                }
                                foreach ($data['tick']['asks'] as $key=>$value)
                                {
                                    $value[0] = $asksprices + (float)$this->randomFloat(1, 5, $jd);
                                    $data['tick']['asks'][$key] = $value;
                                    $asksprices = $value[0];
                                }
                            }
                        }
                    }
                }
                if(isset($this->usersd[$data['ch']])){
                    foreach ($this->usersd[$data['ch']] as $key => $value) {
                        if (!isset($worker->connections[$value])) {
                            unset($this->usersd[$data['ch']][$key]);
                        } else {
                            if(in_array($value,$this->userctrldy)){
                                $datass = $data;
                                $userids = array_search($value, $this->userctrldy);
                                $pricearrs = [];
                                if(isset($this->userctrl[$userids]))
                                {
                                    $tradejson = json_decode($this->userctrl[$userids],true);
                                    if($tradejson) {
                                        foreach ($tradejson as $kk => $vv) {
                                            if ($symbol && $symbol == $kk) {
                                                $jsonarr = explode('|', $vv);
                                                $time1 = explode('-', $jsonarr[0]);
                                                $time2 = explode(':', $time1[0]);
                                                $time3 = explode(':', $time1[1]);
                                                if ($hour >= $time2[0] && $hour <= $time3[0]) {
                                                    if (($hour == $time3[0] && $minute > $time3[1]) || ($hour == $time2[0] && $minute < $time2[1])) continue;
                                                    $pricearrs = explode('-', $jsonarr[1]);
                                                    break;
                                                }
                                            }
                                        }
                                    }
                                }
                                if (!empty($pricearrs)) {
                                    //买一价格
                                    $bidsprice = $pricearrs[0];
                                    $bidsprices = $bidsprice;
                                    //卖一价格
                                    $asksprice = $pricearrs[1];
                                    $asksprices = $asksprice;
                                    foreach ($datass['tick']['bids'] as $kks=>$vals)
                                    {
                                        $vals[0] = $bidsprices - (float)$this->randomFloat(1, 5, $jd);
                                        $datass['tick']['bids'][$kks] = $vals;
                                        $bidsprices = $vals[0];
                                    }
                                    foreach ($datass['tick']['asks'] as $kks=>$vals)
                                    {
                                        $vals[0] = $asksprices + (float)$this->randomFloat(1, 5, $jd);
                                        $datass['tick']['asks'][$kks] = $vals;
                                        $asksprices = $vals[0];
                                    }
                                }
                                $worker->connections[$value]->send(json_encode($datass));
                            }else{
                                $worker->connections[$value]->send(json_encode($data));
                            }
                        }
                    }
                }
                
                //给模拟订阅用户推送
                $ch_depth = explode(".", $data['ch']);
                $release = Db::name("app_curr_release")->where("hb_symbol",$ch_depth[1])->where("status","in","3")->select();
                if($release){
                    foreach ($release as $kkk => $vvvv) {
                        $ch_depth[1] = $vvvv['symbol'];
                        $ch_str = implode(".", $ch_depth);
                        if(isset($this->usersd[$ch_str])){
                            $data['ch'] = $ch_str;
                            foreach ($this->usersd[$ch_str] as $key => $value) {
                                if (!isset($worker->connections[$value])) {
                                    unset($this->usersd[$ch_str][$key]);
                                } else {
                                    $worker->connections[$value]->send(json_encode($data));
                                }
                            }
                        }
                    }
                }
            }
        }else if (isset($data['rep'])){
            $data['subs'] = 'history';
            $data['ch'] = $data['rep'];
            foreach ($data['data'] as $key=>$value)
            {
//                $value['time'] = $value['id'] * 1000;
                $data['data'][$key] = $value;
            }
            $worker->connections[$this->historykline[$data['id']]]->send(json_encode($data));
        }
    }

    function saveLog($symbol, $msg){
        $dir =  __DIR__ ."/logs";
        if( !file_exists($dir) ) mkdir($dir, 0777);
        $today = date('Ymd');
        $file_path =$dir."/a-".$symbol."-".$today.".log";
        $handle = fopen($file_path, "a+");
        @fwrite($handle, date("H:i:s"). $msg . "\r\n");
        @fclose($handle);
    }

    //4位小数的随机数
    function randomFloat($min = 0, $max = 10 , $localnum = 4)
    {
        if($localnum == 4) {
            if ($max - $min <= 0.0002) {
                return 0;
            }
            $rand = mt_rand() / mt_getrandmax() * ($max - $min-0.0001);
            $num = $min + $rand;
            $number = sprintf("%.4f", $num);
            if($number == $min){
                $number += 0.0001;
            }
        }else if($localnum == 5){
            if ($max - $min <= 0.00002) {
                return 0;
            }
            $rand = mt_rand() / mt_getrandmax() * ($max - $min-0.00001);
            $num = $min + $rand;
            $number = sprintf("%.5f", $num);
            if($number == $min){
                $number += 0.00001;
            }
        }else if($localnum == 6){
            if ($max - $min <= 0.000002) {
                return 0;
            }
            $rand = mt_rand() / mt_getrandmax() * ($max - $min-0.000001);
            $num = $min + $rand;
            $number = sprintf("%.6f", $num);
            if($number == $min){
                $number += 0.000001;
            }
        }else if($localnum == 2){
            if ($max - $min <= 0.02) {
                return 0;
            }
            $rand = mt_rand() / mt_getrandmax() * ($max - $min-0.01);
            $num = $min + $rand;
            $number = sprintf("%.2f", $num);
            if($number == $min){
                $number += 0.01;
            }
        }else if($localnum == 3){
            if ($max - $min <= 0.001) {
                return 0;
            }
            $rand = mt_rand() / mt_getrandmax() * ($max - $min-0.001);
            $num = $min + $rand;
            $number = sprintf("%.3f", $num);
            if($number == $min){
                $number += 0.001;
            }
        }

        return $number;

    }
}