<?php
// =============================================================
// COLLECTOR — fetch + normalize fills from Hyperliquid API
// =============================================================

require_once __DIR__ . '/config.php';
require_once __DIR__ . '/logger.php';


// ---------------------------------------------------------------
// Инъекция HTTP-транспорта (DI вместо тестовых швов кэшей)
// ---------------------------------------------------------------
//
// api_post() — единственная точка реального HTTP к Hyperliquid /info для
// одиночных запросов (allMids, clearinghouseState, userFills/ByTime). Если
// установлен транспорт — он вызывается вместо curl. Это даёт тестам подменять
// сеть канонічными ответами без обращения наружу (и без хранения test-only
// функций в прод-коде). В проде транспорт не установлен → идёт реальный curl.
//
// Параллельный батч (fetch_traders_fills_batch, curl_multi) — прод-оптимизация
// и через транспорт НЕ идёт; при сбое он откатывается на fetch_trader_fills →
// api_post, т.е. на транспорт. Тесты дёргают одиночный путь.

/**
 * Возвращает/устанавливает текущий HTTP-транспорт (callable(string $url,
 * string $payload): ?array) или null, если используется реальный curl.
 */
function hl_http_transport(?callable $set = null, bool $clear = false): ?callable
{
    static $fn = null;
    if ($clear)        { $fn = null; }
    elseif ($set !== null) { $fn = $set; }
    return $fn;
}

/** Устанавливает HTTP-транспорт (null — вернуть реальный curl). */
function hl_set_http_transport(?callable $fn): void
{
    if ($fn === null) {
        hl_http_transport(null, true);
    } else {
        hl_http_transport($fn);
    }
}


/**
 * Резолвит хост API один раз за процесс и отдаёт запись для CURLOPT_RESOLVE.
 *
 * Без этого КАЖДЫЙ curl-запрос заново спрашивает DNS «какой адрес у
 * api.hyperliquid.xyz» — при 90+ кошельках это сотни одинаковых DNS-запросов
 * в минуту, за что резолверы хостингов начинают троттлить (симптом:
 * «Resolving timed out» при живой сети). Здесь адрес узнаётся один раз,
 * дальше все запросы процесса используют готовый IP.
 *
 * Если резолв не удался (вернулся не-IP) — возвращает null, и curl резолвит
 * сам, как раньше. Процесс живёт ≤ минуты, устаревание IP не страшно.
 */
function hl_api_resolve_entry(): ?array
{
    static $entry = null;
    static $done  = false;

    if ($done) {
        return $entry;
    }
    $done = true;

    $host = parse_url(HL_API_URL, PHP_URL_HOST);
    if (!is_string($host) || $host === '') {
        return null;
    }

    $scheme = parse_url(HL_API_URL, PHP_URL_SCHEME);
    $port   = parse_url(HL_API_URL, PHP_URL_PORT) ?: ($scheme === 'https' ? 443 : 80);

    $ip = gethostbyname($host);   // при неудаче возвращает сам $host
    if ($ip !== $host && filter_var($ip, FILTER_VALIDATE_IP)) {
        $entry = ["$host:$port:$ip"];
    }

    return $entry;
}

/**
 * Общие curl-опции для POST-запросов к Hyperliquid /info (JSON).
 * Один источник для одиночного api_post() и параллельного
 * fetch_traders_fills_batch() — чтобы заголовки/таймаут не разъезжались.
 */
function hl_api_curl_opts(string $payload): array
{
    $opts = [
        CURLOPT_POST           => true,
        CURLOPT_POSTFIELDS     => $payload,
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_TIMEOUT        => API_TIMEOUT_SEC,
        CURLOPT_HTTPHEADER     => [
            'Content-Type: application/json',
            'Accept: application/json',
        ],
    ];

    $resolve = hl_api_resolve_entry();
    if ($resolve !== null) {
        $opts[CURLOPT_RESOLVE] = $resolve;
    }

    return $opts;
}

/**
 * Получает fills трейдера с Hyperliquid API начиная с указанного времени.
 * Использует userFillsByTime — резко снижает вес запроса (rate limit),
 * так как возвращает только новые fills, а не всю историю.
 *
 * @param string   $wallet     адрес кошелька
 * @param int|null $sinceTsMs  timestamp в мс, с которого брать fills (inclusive).
 *                              null — используется userFills (вся история, только для первого запуска).
 */
function fetch_trader_fills(string $wallet, ?int $sinceTsMs = null, ?bool &$failed = null): array
{
    if ($sinceTsMs === null) {
        // Первый запуск для этого кошелька — берём всю историю одним вызовом
        // только чтобы определить watermark (process_fills сама ничего не исполнит).
        $payload = json_encode([
            'type' => 'userFills',
            'user' => $wallet,
        ]);
    } else {
        $payload = json_encode([
            'type'      => 'userFillsByTime',
            'user'      => $wallet,
            'startTime' => $sinceTsMs,
        ]);
    }

    $result = api_post(HL_API_URL, $payload);

    // $failed отличает сетевую ошибку (все ретраи исчерпаны) от «сделок нет»:
    // обе ситуации возвращают [], но вызывающему коду (fallback в
    // fetch_traders_fills_batch) важна разница — по ошибкам он решает,
    // что сеть лежит, и прекращает опрос остальных кошельков.
    $failed = ($result === null);

    if ($result === null) {
        bot_log('ERROR: failed to fetch fills for ' . $wallet);
        return [];
    }

    return normalize_fills($result);
}

/**
 * Запрашивает общую стоимость аккаунта трейдера (accountValue в USD).
 * Использует clearinghouseState endpoint Hyperliquid.
 * Кэшируется в статической переменной на время одного цикла бота.
 * Возвращает null при ошибке или недоступности данных.
 */
function fetch_trader_equity(string $wallet, bool $reset = false): ?float
{
    static $cache = [];

    if ($reset) {
        $cache = [];
        return null;
    }

    if (array_key_exists($wallet, $cache)) {
        return $cache[$wallet];
    }

    $payload = json_encode([
        'type' => 'clearinghouseState',
        'user' => $wallet,
    ]);

    $result = api_post(HL_API_URL, $payload);

    if ($result === null || !isset($result['marginSummary']['accountValue'])) {
        bot_log('PROPORTIONAL: не удалось получить equity для ' . $wallet);
        $cache[$wallet] = null;
        return null;
    }

    $equity = (float) $result['marginSummary']['accountValue'];
    $cache[$wallet] = $equity;

    return $equity;
}

/**
 * Сбрасывает кэш equity трейдеров между итерациями цикла.
 * Вызывается в начале каждой итерации в bot.php вместе с fetch_reset_price_cache()
 * — иначе все циклы в рамках одного процесса видели бы equity первой итерации.
 */
function fetch_reset_trader_equity_cache(): void
{
    fetch_trader_equity('', true);
}

/**
 * Нормализует ответ API в единый формат.
 * Автоматически фильтрует спот и стейкинг — только фьючерсы (perp).
 *
 * Частичные исполнения одного ордера (один oid, разные tid) АГРЕГИРУЮТСЯ в
 * ОДИН fill: размер суммируется, цена — объёмно-взвешенная (VWAP), время —
 * последнего исполнения. Раньше дедуп по oid оставлял лишь ПЕРВЫЙ кусок ордера,
 * из-за чего trader_size_coins/usd занижались (особенно бьёт по доле частичного
 * закрытия в partial_close_mode и по proportional sizing). Агрегация работает в
 * пределах одного ответа API; куски, пришедшие в РАЗНЫХ циклах, по-прежнему
 * схлопываются дедупом seen_fills в первый кусок (полная кросс-цикловая склейка
 * потребовала бы хранить частичные суммы в state — намеренно не делаем).
 */
function normalize_fills(array $raw): array
{
    $byOid = [];   // oid => агрегат
    $order = [];   // порядок первого появления oid (сохраняем хронологию ответа)

    foreach ($raw as $item) {
        // Пропускаем если нет обязательных полей
        if (empty($item['oid']) || empty($item['coin']) || empty($item['side']) || empty($item['sz'])) {
            continue;
        }

        $coin = $item['coin'];

        // Фильтр фьючерсов:
        // Споты на Hyperliquid: "@107", "BTC/USDC" — содержат @ или /
        // Фьючерсы (perp): чистый тикер "BTC", "ETH", "SOL"
        // Стейкинг: не попадает в userFills, но стейблы фильтруем на всякий случай
        if (!is_futures_coin($coin)) {
            bot_log('SKIP non-futures coin: ' . $coin);
            continue;
        }

        $oid  = (string) $item['oid'];
        $sz   = (float) $item['sz'];
        $px   = isset($item['px']) ? (float) $item['px'] : null;
        $time = isset($item['time']) ? (int) $item['time'] : time() * 1000;

        if (!isset($byOid[$oid])) {
            $byOid[$oid] = [
                // id = oid (order id). Копитрейдинг реагирует на ОРДЕР, а не на
                // каждое частичное исполнение (tid) — поэтому ключ агрегации oid.
                'id'           => $oid,
                'symbol'       => strtoupper($coin),
                // Биржевое имя КАК ЕСТЬ ('kPEPE'): symbol нормализован к верхнему
                // регистру для внутреннего учёта, но Hyperliquid-нативным приёмникам
                // (katoshi) нужен точный регистр биржи — см. send_katoshi_webhook.
                'symbol_exchange' => $coin,
                'side'         => $item['side'],   // 'B' (Buy/Long) или 'A' (Ask/Short)
                // dir — намерение трейдера: "Open Long"/"Close Long"/реверс. У всех
                // кусков одного ордера одинаков; берём первый известный.
                'dir'          => isset($item['dir']) ? (string) $item['dir'] : null,
                'size'         => 0.0,
                'trader_price' => null,
                'timestamp'    => $time,
                '_pxNotional'  => 0.0,   // Σ(sz·px) — числитель VWAP
                '_pxSize'      => 0.0,   // Σ(sz) по кускам с известной px — знаменатель VWAP
            ];
            $order[] = $oid;
        }

        $byOid[$oid]['size'] += $sz;
        if ($px !== null) {
            $byOid[$oid]['_pxNotional'] += $sz * $px;
            $byOid[$oid]['_pxSize']     += $sz;
        }
        if ($time > $byOid[$oid]['timestamp']) {
            $byOid[$oid]['timestamp'] = $time;   // время последнего исполнения ордера
        }
        if ($byOid[$oid]['dir'] === null && isset($item['dir'])) {
            $byOid[$oid]['dir'] = (string) $item['dir'];
        }
    }

    $fills = [];
    foreach ($order as $oid) {
        $agg = $byOid[$oid];
        // VWAP по кускам с известной ценой; если цены не было ни у одного — null.
        $agg['trader_price'] = $agg['_pxSize'] > 0 ? $agg['_pxNotional'] / $agg['_pxSize'] : null;
        unset($agg['_pxNotional'], $agg['_pxSize']);
        $fills[] = $agg;
    }

    return $fills;
}

/**
 * Проверяет что монета является фьючерсным (perp) инструментом.
 */
function is_futures_coin(string $coin): bool
{
    // Спот токены по числовому ID: @1, @107 и т.д.
    if (strpos($coin, '@') !== false) return false;

    // Спот пары: BTC/USDC
    if (strpos($coin, '/') !== false) return false;

    // Стейблы — не являются perp инструментами
    $stables = ['USDC', 'USDT', 'DAI', 'BUSD', 'FDUSD'];
    if (in_array(strtoupper($coin), $stables, true)) return false;

    return true;
}


/**
 * Получает fills для группы кошельков ПАРАЛЛЕЛЬНО через curl_multi.
 *
 * У Hyperliquid НЕТ multi-user эндпоинта (запрос на массив `/info` не
 * существует), поэтому это по одному запросу userFills(ByTime) на кошелёк, но
 * выполненных одновременно — суммарное время ≈ времени самого медленного
 * запроса, а не их сумме. Кошельки без watermark (первый запуск) идут как
 * userFills (вся история), остальные — userFillsByTime от watermark.
 *
 * Отказоустойчивость: запросы, не удавшиеся в параллельном проходе (curl error,
 * не-200, битый JSON), до-запрашиваются по одному через fetch_trader_fills() —
 * там есть retry с экспоненциальным backoff. Так быстрый happy-path сочетается с
 * надёжным fallback без дублирования логики ретраев.
 *
 * Замечание про rate-limit: параллельный всплеск из N запросов укладывается в
 * weight-бюджет Hyperliquid для разумных N; при больших списках кошельков число
 * запросов за цикл ограничивается WALLET_BATCH_SIZE (см. bot.php).
 *
 * @param array      $wallets    массив ['walletId' => ['address' => ..., 'sinceTsMs' => int|null]]
 * @param array|null $failedIds  out-параметр: walletId, чей fetch НЕ удался даже
 *                               с ретраями (сетевой сбой, а не «сделок нет»).
 *                               Вызывающий код не должен для них двигать
 *                               watermark/wallet_sync — fills доберутся позже.
 * @return array                 массив ['walletId' => [fills...]]
 */
function fetch_traders_fills_batch(array $wallets, ?array &$failedIds = null): array
{
    $failedIds = [];

    if (empty($wallets)) {
        return [];
    }

    // Один кошелёк — параллелить нечего, обычный путь с retry+backoff.
    if (count($wallets) === 1) {
        $walletId    = array_key_first($wallets);
        $meta        = $wallets[$walletId];
        $fetchFailed = false;
        $fills       = fetch_trader_fills($meta['address'], $meta['sinceTsMs'], $fetchFailed);
        if ($fetchFailed) {
            $failedIds[] = $walletId;
        }
        return [$walletId => $fills];
    }

    $mh      = curl_multi_init();
    $handles = []; // walletId => curl handle

    foreach ($wallets as $walletId => $meta) {
        $payload = $meta['sinceTsMs'] === null
            ? json_encode(['type' => 'userFills', 'user' => $meta['address']])
            : json_encode(['type' => 'userFillsByTime', 'user' => $meta['address'], 'startTime' => $meta['sinceTsMs']]);

        $ch = curl_init(HL_API_URL);
        curl_setopt_array($ch, hl_api_curl_opts($payload));
        curl_multi_add_handle($mh, $ch);
        $handles[$walletId] = $ch;
    }

    // Крутим пул, пока все запросы не завершатся.
    do {
        $status = curl_multi_exec($mh, $running);
        if ($running) {
            // Блокируемся до активности на сокетах; -1 = нет fd, чуть подождём,
            // чтобы не крутить busy-loop.
            if (curl_multi_select($mh, 1.0) === -1) {
                usleep(1000);
            }
        }
    } while ($running && $status === CURLM_OK);

    $result = [];
    $failed = [];

    foreach ($handles as $walletId => $ch) {
        $response  = curl_multi_getcontent($ch);
        $httpCode  = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $curlError = curl_error($ch);
        curl_multi_remove_handle($mh, $ch);
        curl_close($ch);

        if ($curlError || $httpCode !== 200) {
            bot_log(sprintf('PARALLEL fetch failed for %s (HTTP %d %s) — retrying singly', $walletId, $httpCode, $curlError));
            $failed[$walletId] = $wallets[$walletId];
            continue;
        }

        $decoded = json_decode($response, true);
        if (!is_array($decoded)) {
            bot_log('PARALLEL fetch: invalid JSON for ' . $walletId . ' — retrying singly');
            $failed[$walletId] = $wallets[$walletId];
            continue;
        }

        $result[$walletId] = normalize_fills($decoded);
    }

    curl_multi_close($mh);

    // Если упали ВСЕ кошельки разом — это не точечные сбои, а лежащая сеть
    // (DNS/связность). Последовательный до-запрос в этом состоянии бесполезен
    // и растягивает цикл на ~10с × N кошельков (при 90+ кошельках — на десятки
    // минут), из-за чего запуски наслаиваются друг на друга. Пропускаем цикл:
    // watermark не двигается, fills доберутся следующим запуском.
    if (empty($result) && count($failed) === count($wallets)) {
        bot_log(sprintf(
            'PARALLEL fetch: all %d wallet(s) failed — network looks down, skipping per-wallet retries this cycle',
            count($failed)
        ));
        $failedIds = array_keys($failed);
        return array_map(fn() => [], $failed);
    }

    // До-запрос неудавшихся по одному (fetch_trader_fills ретраит с backoff).
    //
    // Предохранитель: при мерцающей сети (DNS то жив, то нет) параллельный
    // проход может частично удаться, и тогда сюда прилетают ПОЧТИ все кошельки.
    // Каждый до-запрос при лежащей сети — это ~10с таймаутов; на 90 кошельках
    // цикл растягивался на десятки минут. Три сетевые ошибки ПОДРЯД — считаем,
    // что сеть лежит, и оставшихся не опрашиваем (fills доберутся следующим
    // циклом: watermark не двигается). Успешный ответ сбрасывает счётчик.
    $maxConsecutiveFailures = 3;
    $consecutiveFailures    = 0;

    foreach ($failed as $walletId => $meta) {
        if ($consecutiveFailures >= $maxConsecutiveFailures) {
            $result[$walletId]  = [];
            $failedIds[] = $walletId;   // не опрошен — сбой, а не «сделок нет»
            continue;
        }

        $fetchFailed = false;
        $result[$walletId] = fetch_trader_fills($meta['address'], $meta['sinceTsMs'], $fetchFailed);

        if (!$fetchFailed) {
            $consecutiveFailures = 0;
            continue;
        }

        $failedIds[] = $walletId;
        $consecutiveFailures++;
        if ($consecutiveFailures >= $maxConsecutiveFailures) {
            bot_log(sprintf(
                'FALLBACK fetch: %d consecutive network failures — network looks down, skipping remaining wallet(s) this cycle',
                $consecutiveFailures
            ));
        }
    }

    return $result;
}

/**
 * Экспоненциальный backoff между ретраями API.
 * Без него цикл ретраев бил по API мгновенно (особенно плохо при 429).
 * Не спит после последней попытки (когда повторять уже не будем).
 */
function api_backoff_sleep(int $attempt, bool $rateLimited = false): void
{
    if ($attempt >= (API_RETRY_COUNT + 1)) {
        return; // это была последняя попытка — спать незачем
    }
    $baseMs = $rateLimited ? 1000 : 200;
    usleep($baseMs * 1000 * $attempt); // 200,400,... мс (или 1000,2000,... при 429)
}

/**
 * HTTP POST к API с retry.
 * Возвращает декодированный массив или null при ошибке.
 */
function api_post(string $url, string $payload): ?array
{
    // Инъектированный транспорт (тесты) — вместо реального curl.
    $transport = hl_http_transport();
    if ($transport !== null) {
        return $transport($url, $payload);
    }

    $attempts = 0;
    $maxAttempts = API_RETRY_COUNT + 1;

    while ($attempts < $maxAttempts) {
        $ch = curl_init($url);
        curl_setopt_array($ch, hl_api_curl_opts($payload));

        $response = curl_exec($ch);
        $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $curlError = curl_error($ch);
        curl_close($ch);

        if ($curlError) {
            bot_log('CURL error (attempt ' . ($attempts + 1) . '): ' . $curlError);
            $attempts++;
            api_backoff_sleep($attempts);
            continue;
        }

        if ($httpCode !== 200) {
            bot_log('HTTP ' . $httpCode . ' from API (attempt ' . ($attempts + 1) . ')');
            $attempts++;
            // 429 (rate limit) ждёт дольше, чтобы не усугублять упор в лимит веса.
            api_backoff_sleep($attempts, $httpCode === 429);
            continue;
        }

        $decoded = json_decode($response, true);
        if (!is_array($decoded)) {
            bot_log('Invalid JSON from API');
            $attempts++;
            api_backoff_sleep($attempts);
            continue;
        }

        return $decoded;
    }

    return null;
}
