li
1b24994e74
- Contract.php: 返回合约账户余额(balance_contract) - My.php: 地址管理增加BTC/ETH - AppContract.php: 一键平仓(closeall) - AppProxy.php: 代理专属注册链接 + 分级权限(L1/L2) - site.php: 手续费减半(0.018→0.009) - agent_permission_setup.sql: 代理权限SQL - crypto_news_crawler.py: 新闻自动采集脚本
968 lines
42 KiB
PHP
Executable File
968 lines
42 KiB
PHP
Executable File
<?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;
|
||
|
||
}
|
||
} |