SYSTEM_CORE_VIEWER
DAEMON
Target Object / Identifier
daemon_upbit_Ticker.php
업비트 플랫폼
업비트 마켓 API --> 플랫폼 실시간
Exchange
업비트
TABLE
daemon_upbit_Ticker
Daemon Process
daemon_upbit_Ticker.php
TYPE
플랫폼
FORM
실시간
MANAGEMENT
START
[STATUS] ACTIVE
[ID] #1
[DAEMON] daemon_upbit_Ticker.php
[TABLE] daemon_upbit_Ticker
Detailed Description
* 업비트 API 마켓 매매 데이터 수집 데몬
1. 업비트 API 마켓 매매 테이터를 수집 테이블에 저장한다.
가. 데몬 이름 : daemon_upbit_Ticker.php
나. 수집 테이블 : daemon_upbit_Ticker
다. 수집 주기 : 초당
1. 업비트 API 마켓 매매 테이터를 수집 테이블에 저장한다.
가. 데몬 이름 : daemon_upbit_Ticker.php
나. 수집 테이블 : daemon_upbit_Ticker
다. 수집 주기 : 초당
Daemon Source Code (daemon_upbit_Ticker.php)
#!/usr/bin/php
<?php
/**
* ============================================================
* 업비트 전체 시세 수집 CLI 데몬
* - curl_multi 병렬 호출 방식
* - gnu DB 루프 밖 1회 연결
* - bulk INSERT 한방 쿼리
* - echo/flush 전부 제거
* ============================================================
*/
error_reporting(E_ALL);
ini_set('display_errors', 1);
date_default_timezone_set('Asia/Seoul');
$DAEMON_ID = pathinfo(__FILE__, PATHINFO_FILENAME);
if (php_sapi_name() !== 'cli') {
exit;
}
function get_db_connection() {
try {
$db_upbit = null;
@include '/home/www/DB/db_upbit.php';
if (!($db_upbit instanceof PDO)) return null;
$pdo = $db_upbit;
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
return $pdo;
} catch (Throwable $e) {
return null;
}
}
function get_gnu_connection() {
try {
$db_gnu = null;
$pdo_gnu = null;
$pdo = null;
@include '/home/www/DB/db_gnu.php';
if ($db_gnu instanceof PDO) {
$db_gnu->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
return $db_gnu;
}
if ($pdo_gnu instanceof PDO) {
$pdo_gnu->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
return $pdo_gnu;
}
if ($pdo instanceof PDO) {
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
return $pdo;
}
return null;
} catch (Throwable $e) {
return null;
}
}
// ============================================================
// curl_multi 병렬 호출
// ============================================================
function http_multi_get(array $urls): array {
$mh = curl_multi_init();
$handles = [];
foreach ($urls as $key => $url) {
$ch = curl_init($url);
curl_setopt_array($ch, [
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => 8,
CURLOPT_USERAGENT => 'upbit-ghost',
CURLOPT_SSL_VERIFYPEER => false,
]);
curl_multi_add_handle($mh, $ch);
$handles[$key] = $ch;
}
$running = null;
do {
curl_multi_exec($mh, $running);
curl_multi_select($mh);
} while ($running > 0);
$results = [];
foreach ($handles as $key => $ch) {
$raw = curl_multi_getcontent($ch);
$results[$key] = $raw ? json_decode($raw, true) : null;
curl_multi_remove_handle($mh, $ch);
curl_close($ch);
}
curl_multi_close($mh);
return $results;
}
$api = require '/home/www/DB/upbit_api_url.php';
$API_TICKER = $api['ticker'] . '?markets=';
$API_ORDERBOOK = 'https://api.upbit.com/v1/orderbook?markets=';
$API_TRADES = 'https://api.upbit.com/v1/trades/ticks?count=1&market=';
$pdo = get_db_connection();
$pdo_gnu = get_gnu_connection();
$server_ip = 'CLI_DAEMON';
$cycle_count = 0;
$last_market_refresh = 0;
$krw = [];
$stmt_hb = null;
$stmt_kill = null;
$stmt_stop = null;
$stmt_best = null;
while (true) {
$cycle_count++;
try {
$reconnected = false;
if (!$pdo) {
$pdo = get_db_connection();
$reconnected = true;
} else {
try { $pdo->query("SELECT 1"); }
catch (Throwable $e) {
$pdo = get_db_connection();
$reconnected = true;
}
}
if ($reconnected) {
$stmt_hb = $stmt_kill = $stmt_stop = $stmt_best = null;
}
if ($pdo) {
if (!$stmt_hb) {
$stmt_hb = $pdo->prepare("
INSERT INTO daemon_record (
d_id, d_category, d_pid, d_status,
d_heartbeat, d_ip, d_start_time, d_memo, d_kill_flag
)
VALUES (
:id, 'UPBIT', :pid, 'RUNNING',
NOW(), :ip, NOW(), 'UPBIT BOARD ONLY TICKER GHOST', 0
)
ON DUPLICATE KEY UPDATE
d_pid = VALUES(d_pid),
d_status = 'RUNNING',
d_heartbeat = NOW(),
d_ip = VALUES(d_ip),
d_memo = VALUES(d_memo)
");
}
if (!$stmt_kill) {
$stmt_kill = $pdo->prepare("SELECT d_kill_flag FROM daemon_record WHERE d_id = :id LIMIT 1");
}
if (!$stmt_stop) {
$stmt_stop = $pdo->prepare("
UPDATE daemon_record
SET d_status='STOPPED', d_heartbeat=NOW(), d_pid=0
WHERE d_id=:id
");
}
$stmt_hb->execute([':id' => $DAEMON_ID, ':pid' => getmypid(), ':ip' => $server_ip]);
$stmt_kill->execute([':id' => $DAEMON_ID]);
$kill_flag = (int)($stmt_kill->fetchColumn() ?: 0);
if ($kill_flag === 1) {
$stmt_stop->execute([':id' => $DAEMON_ID]);
exit(0);
}
// 종목 갱신 (1초마다)
if (time() - $last_market_refresh >= 1 || empty($krw)) {
$tmp = [];
if (!$pdo_gnu) {
$pdo_gnu = get_gnu_connection();
$stmt_best = null;
} else {
try { $pdo_gnu->query("SELECT 1"); }
catch (Throwable $e) {
$pdo_gnu = get_gnu_connection();
$stmt_best = null;
}
}
if ($pdo_gnu instanceof PDO) {
if (!$stmt_best) {
$stmt_best = $pdo_gnu->prepare("
SELECT wr_subject FROM g5_write_daemon_kind_upbit
WHERE (x2_run = 1 OR x2_run = '1')
");
}
$stmt_best->execute();
$best_rows = $stmt_best->fetchAll(PDO::FETCH_COLUMN);
if (is_array($best_rows)) {
foreach ($best_rows as $sym) {
$sym = strtoupper(trim((string)$sym));
if ($sym === '') continue;
if (strpos($sym, '-') === false) {
$sym = 'KRW-' . $sym;
}
if (strpos($sym, 'KRW-') !== 0) continue;
if (!in_array($sym, $tmp, true)) $tmp[] = $sym;
}
}
}
if ($tmp) {
$krw = $tmp;
$last_market_refresh = time();
} else {
$krw = [];
sleep(10);
continue;
}
}
if (!$krw) {
sleep(10);
continue;
}
$at = date('Y-m-d H:i:s');
$ms = (int)(microtime(true) * 1000);
// 티커 + 호가 동시 호출, trades는 종목별 개별 URL
$market_str = implode(',', $krw);
$multi_urls = [
'ticker' => $API_TICKER . $market_str,
'orderbook' => $API_ORDERBOOK . $market_str,
];
foreach ($krw as $sym) {
$multi_urls["trades__{$sym}"] = $API_TRADES . urlencode($sym);
}
$raw = http_multi_get($multi_urls);
$tks = $raw['ticker'];
$obs = $raw['orderbook'];
// trades 맵 구성
$tr_map = [];
foreach ($krw as $sym) {
$tr_res = $raw["trades__{$sym}"] ?? null;
if (is_array($tr_res) && isset($tr_res[0])) {
$tr_map[$sym] = $tr_res[0];
}
}
if (!$tks || !is_array($tks)) {
sleep(3);
continue;
}
// 호가 맵 구성
$ob_map = [];
if (is_array($obs)) {
foreach ($obs as $o) {
if (isset($o['market'])) {
$ob_map[$o['market']] = $o;
}
}
}
// bulk INSERT
$placeholders = [];
$params = [];
foreach ($tks as $i => $t) {
$market = $t['market'] ?? '';
if ($market === '') continue;
$ob = $ob_map[$market] ?? [];
$tr = $tr_map[$market] ?? [];
$placeholders[] = "(
:market_{$i}, :trade_date_{$i}, :trade_time_{$i}, :trade_date_kst_{$i}, :trade_time_kst_{$i},
:opening_price_{$i}, :high_price_{$i}, :low_price_{$i}, :trade_price_{$i}, :prev_closing_price_{$i},
:change_{$i}, :change_price_{$i}, :change_rate_{$i}, :signed_change_price_{$i}, :signed_change_rate_{$i},
:trade_volume_{$i}, :acc_trade_volume_{$i}, :acc_trade_volume_24h_{$i}, :acc_trade_price_{$i}, :acc_trade_price_24h_{$i},
:highest_52_week_price_{$i}, :highest_52_week_date_{$i}, :lowest_52_week_price_{$i}, :lowest_52_week_date_{$i},
:collected_at_{$i}, :collected_ms_{$i},
:ob_timestamp_{$i}, :ob_total_ask_size_{$i}, :ob_total_bid_size_{$i}, :ob_units_{$i},
:ob_collected_at_{$i}, :ob_collected_ms_{$i},
:tr_trade_timestamp_{$i}, :tr_trade_price_{$i}, :tr_trade_volume_{$i}, :tr_ask_bid_{$i},
:tr_trade_date_utc_{$i}, :tr_trade_time_utc_{$i}, :tr_trade_date_kst_{$i}, :tr_trade_time_kst_{$i},
:tr_collected_at_{$i}, :tr_collected_ms_{$i},
:day_of_week_{$i}, :korean_name_{$i}
)";
$params["market_{$i}"] = $market;
$params["trade_date_{$i}"] = $t['trade_date'] ?? '';
$params["trade_time_{$i}"] = $t['trade_time'] ?? '';
$params["trade_date_kst_{$i}"] = $t['trade_date_kst'] ?? '';
$params["trade_time_kst_{$i}"] = $t['trade_time_kst'] ?? '';
$params["opening_price_{$i}"] = $t['opening_price'] ?? 0;
$params["high_price_{$i}"] = $t['high_price'] ?? 0;
$params["low_price_{$i}"] = $t['low_price'] ?? 0;
$params["trade_price_{$i}"] = $t['trade_price'] ?? 0;
$params["prev_closing_price_{$i}"] = $t['prev_closing_price'] ?? 0;
$params["change_{$i}"] = $t['change'] ?? '';
$params["change_price_{$i}"] = $t['change_price'] ?? 0;
$params["change_rate_{$i}"] = $t['change_rate'] ?? 0;
$params["signed_change_price_{$i}"] = $t['signed_change_price'] ?? 0;
$params["signed_change_rate_{$i}"] = $t['signed_change_rate'] ?? 0;
$params["trade_volume_{$i}"] = $t['trade_volume'] ?? 0;
$params["acc_trade_volume_{$i}"] = $t['acc_trade_volume'] ?? 0;
$params["acc_trade_volume_24h_{$i}"] = $t['acc_trade_volume_24h'] ?? 0;
$params["acc_trade_price_{$i}"] = $t['acc_trade_price'] ?? 0;
$params["acc_trade_price_24h_{$i}"] = $t['acc_trade_price_24h'] ?? 0;
$params["highest_52_week_price_{$i}"] = $t['highest_52_week_price'] ?? 0;
$params["highest_52_week_date_{$i}"] = $t['highest_52_week_date'] ?? '';
$params["lowest_52_week_price_{$i}"] = $t['lowest_52_week_price'] ?? 0;
$params["lowest_52_week_date_{$i}"] = $t['lowest_52_week_date'] ?? '';
$params["collected_at_{$i}"] = $at;
$params["collected_ms_{$i}"] = $ms;
$params["tr_trade_timestamp_{$i}"] = $tr['timestamp'] ?? 0;
$params["tr_trade_price_{$i}"] = $tr['trade_price'] ?? 0;
$params["tr_trade_volume_{$i}"] = $tr['trade_volume'] ?? 0;
$params["tr_ask_bid_{$i}"] = $tr['ask_bid'] ?? '';
$params["tr_trade_date_utc_{$i}"] = $tr['trade_date_utc'] ?? '';
$params["tr_trade_time_utc_{$i}"] = $tr['trade_time_utc'] ?? '';
$params["tr_trade_date_kst_{$i}"] = $t['trade_date_kst'] ?? '';
$params["tr_trade_time_kst_{$i}"] = $t['trade_time_kst'] ?? '';
$params["tr_collected_at_{$i}"] = $at;
$params["tr_collected_ms_{$i}"] = $ms;
$params["ob_timestamp_{$i}"] = $ob['timestamp'] ?? 0;
$params["ob_total_ask_size_{$i}"] = $ob['total_ask_size'] ?? 0;
$params["ob_total_bid_size_{$i}"] = $ob['total_bid_size'] ?? 0;
$params["ob_units_{$i}"] = isset($ob['orderbook_units'])
? json_encode($ob['orderbook_units'], JSON_UNESCAPED_UNICODE)
: null;
$params["ob_collected_at_{$i}"] = $at;
$params["ob_collected_ms_{$i}"] = $ms;
$params["day_of_week_{$i}"] = (int)date('w');
$params["korean_name_{$i}"] = '';
}
if (!empty($placeholders)) {
$sql = "
INSERT INTO daemon_upbit_Ticker (
market, trade_date, trade_time, trade_date_kst, trade_time_kst,
opening_price, high_price, low_price, trade_price, prev_closing_price,
`change`, change_price, change_rate, signed_change_price, signed_change_rate,
trade_volume, acc_trade_volume, acc_trade_volume_24h, acc_trade_price, acc_trade_price_24h,
highest_52_week_price, highest_52_week_date, lowest_52_week_price, lowest_52_week_date,
collected_at, collected_ms,
ob_timestamp, ob_total_ask_size, ob_total_bid_size, ob_units,
ob_collected_at, ob_collected_ms,
tr_trade_timestamp, tr_trade_price, tr_trade_volume, tr_ask_bid,
tr_trade_date_utc, tr_trade_time_utc, tr_trade_date_kst, tr_trade_time_kst,
tr_collected_at, tr_collected_ms,
day_of_week, korean_name
) VALUES " . implode(',', $placeholders) . "
ON DUPLICATE KEY UPDATE
trade_date = VALUES(trade_date),
trade_time = VALUES(trade_time),
trade_date_kst = VALUES(trade_date_kst),
trade_time_kst = VALUES(trade_time_kst),
opening_price = VALUES(opening_price),
high_price = VALUES(high_price),
low_price = VALUES(low_price),
trade_price = VALUES(trade_price),
prev_closing_price = VALUES(prev_closing_price),
`change` = VALUES(`change`),
change_price = VALUES(change_price),
change_rate = VALUES(change_rate),
signed_change_price = VALUES(signed_change_price),
signed_change_rate = VALUES(signed_change_rate),
trade_volume = VALUES(trade_volume),
acc_trade_volume = VALUES(acc_trade_volume),
acc_trade_volume_24h = VALUES(acc_trade_volume_24h),
acc_trade_price = VALUES(acc_trade_price),
acc_trade_price_24h = VALUES(acc_trade_price_24h),
highest_52_week_price = VALUES(highest_52_week_price),
highest_52_week_date = VALUES(highest_52_week_date),
lowest_52_week_price = VALUES(lowest_52_week_price),
lowest_52_week_date = VALUES(lowest_52_week_date),
collected_at = VALUES(collected_at),
collected_ms = VALUES(collected_ms),
tr_trade_timestamp = VALUES(tr_trade_timestamp),
tr_trade_price = VALUES(tr_trade_price),
tr_trade_volume = VALUES(tr_trade_volume),
tr_ask_bid = VALUES(tr_ask_bid),
tr_trade_date_utc = VALUES(tr_trade_date_utc),
tr_trade_time_utc = VALUES(tr_trade_time_utc),
tr_trade_date_kst = VALUES(tr_trade_date_kst),
tr_trade_time_kst = VALUES(tr_trade_time_kst),
tr_collected_at = VALUES(tr_collected_at),
tr_collected_ms = VALUES(tr_collected_ms),
ob_timestamp = VALUES(ob_timestamp),
ob_total_ask_size = VALUES(ob_total_ask_size),
ob_total_bid_size = VALUES(ob_total_bid_size),
ob_units = VALUES(ob_units),
ob_collected_at = VALUES(ob_collected_at),
ob_collected_ms = VALUES(ob_collected_ms),
day_of_week = VALUES(day_of_week),
korean_name = VALUES(korean_name)
";
$stmt = $pdo->prepare($sql);
$stmt->execute($params);
}
} else {
sleep(5);
}
if ($cycle_count % 50 === 0 && function_exists('gc_collect_cycles')) {
gc_collect_cycles();
}
} catch (Throwable $e) {
sleep(3);
}
sleep(30);
}