<?php
/*
|--------------------------------------------------------------------------
| sync_candles.php
|--------------------------------------------------------------------------
| Continuous candle sync worker for cron.
|
| Purpose:
| - Initial history/backfill: php sync_candles.php backfill --symbols=BTCUSD --timeframes=5M,15M,1H,4H,1D --limit=300
| - Live updates:          php sync_candles.php live --asset-types=CRYPTO --symbols=BTCUSD
|
| Recommended cron:
| Every 5 minutes: /usr/bin/php /path/to/app/cron/sync_candles.php live >/dev/null 2>&1
|
| Notes:
| - This worker updates candles only. Alerts are still handled by run_institutional_alerts.php.
| - It prefers active symbols from the symbols table/SystemConfigService.
| - It can work even if some timeframes are not yet in the timeframes table because it has a built-in interval map.
|--------------------------------------------------------------------------
*/

ini_set('display_errors', 1);
error_reporting(E_ALL);
date_default_timezone_set('UTC');

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

if (file_exists(__DIR__ . '/../services/SystemConfigService.php')) {
    require_once __DIR__ . '/../services/SystemConfigService.php';
}
if (file_exists(__DIR__ . '/../services/MarketDataService.php')) {
    require_once __DIR__ . '/../services/MarketDataService.php';
}
if (file_exists(__DIR__ . '/../services/ApiCreditGuard.php')) {
    require_once __DIR__ . '/../services/ApiCreditGuard.php';
}
if (file_exists(__DIR__ . '/../services/ApiProviderRotationService.php')) {
    require_once __DIR__ . '/../services/ApiProviderRotationService.php';
}
if (file_exists(__DIR__ . '/../services/CandleService.php')) {
    require_once __DIR__ . '/../services/CandleService.php';
}

const SYNC_CANDLES_BUILD = 'SYNC_CANDLES_ENTERPRISE_QUEUE_V1_2026_06_06';

$argvList = $_SERVER['argv'] ?? [];
$mode = 'live';

if (isset($argvList[1]) && strpos((string)$argvList[1], '--') !== 0) {
    $mode = strtolower(trim((string)$argvList[1]));
}

$options = sync_cli_options($argvList);

if (isset($options['mode'])) {
    $mode = strtolower(trim((string)$options['mode']));
}

if (!in_array($mode, ['live', 'backfill', 'test', 'dry-run'], true)) {
    $mode = 'live';
}

$dryRun = $mode === 'dry-run' || sync_bool_option($options, 'dry-run', false);
$coreOnly = sync_bool_option($options, 'core-only', true);
$includeInactive = sync_bool_option($options, 'include-inactive', false);
$providerOverride = strtolower(trim((string)($options['provider'] ?? '')));
$outputLimit = isset($options['limit']) ? max(1, min(1000, (int)$options['limit'])) : null;
$maxSymbols = isset($options['max-symbols']) ? max(1, (int)$options['max-symbols']) : 0;
$maxPairs = isset($options['max-pairs']) ? max(1, (int)$options['max-pairs']) : 0;
$force = sync_bool_option($options, 'force', false);
$assetTypeFilter = sync_requested_asset_types($options);

/*
 * Alert-aware production mode:
 * When enabled, the worker does not sync every active symbol/timeframe.
 * It only syncs the symbols and exact timeframes needed by active tracked setups.
 */
$explicitSymbolsRequested = trim((string)($options['symbols'] ?? $options['symbol'] ?? '')) !== '';
$alertAware = sync_bool_option($options, 'alert-aware', false)
    || sync_bool_option($options, 'active-alerts-only', false)
    || sync_bool_option($options, 'tracked-only', false)
    || (
        $mode === 'live'
        && !$explicitSymbolsRequested
        && sync_setting_enabled('candle_sync_alert_aware_default', true)
    );

/*
 * Enterprise scheduler mode:
 * For large platforms, do not let one cron scan every active setup directly.
 * The scheduler builds a symbol+timeframe queue, ranks jobs by plan/stage,
 * locks only a controlled number of due jobs, then processes those jobs.
 * This keeps the same cron usable on a small API plan while the data model is
 * ready for multiple workers later.
 */
$enterpriseScheduler = $alertAware && sync_setting_enabled('candle_sync_scheduler_enabled', false);
$queueWorkerId = 'cw_' . preg_replace('/[^A-Za-z0-9_\-]/', '_', (string)(gethostname() ?: 'host')) . '_' . getmypid() . '_' . substr(md5((string)microtime(true)), 0, 8);
$queueBuildSummary = null;

$allowPaidCandles = sync_bool_option($options, 'allow-paid-candles', false)
    || sync_bool_option($options, 'allow-paid', false)
    || (
        $alertAware
        && sync_setting_enabled('candle_sync_alert_aware_allow_paid', true)
    );

$defaultLiveLimit = (int)sync_setting('candle_sync_live_limit', '10');
$defaultBackfillLimit = (int)sync_setting('candle_sync_backfill_limit', '300');

if ($outputLimit === null) {
    $outputLimit = $mode === 'backfill' ? $defaultBackfillLimit : $defaultLiveLimit;
}

$outputLimit = max(1, min(1000, $outputLimit));

$timeframes = sync_requested_timeframes($options);

if ($alertAware && $enterpriseScheduler) {
    sync_ensure_candle_queue_tables();
    $queueBuildSummary = sync_rebuild_alert_aware_queue($options, $timeframes);
    $queueLimit = isset($options['queue-limit'])
        ? max(1, min(10000, (int)$options['queue-limit']))
        : (int)sync_setting('candle_sync_scheduler_jobs_per_run', (string)sync_setting('candle_sync_alert_aware_max_distinct_pairs', '50'));
    $queueLimit = max(1, min(10000, $queueLimit));
    if ($maxPairs > 0) {
        $queueLimit = min($queueLimit, $maxPairs);
    }
    $symbols = sync_queue_due_symbol_rows($queueWorkerId, $queueLimit, $options);
} elseif ($alertAware) {
    $symbols = sync_active_alert_symbol_rows($options, $timeframes);
} else {
    $symbols = sync_requested_symbols($options, $coreOnly, $includeInactive);
}

/*
 * Final defensive dedupe. Even if many users track the same symbol/timeframe,
 * candle sync must process each symbol/timeframe once per run, not once per setup.
 */
$symbols = sync_deduplicate_symbol_rows($symbols, $timeframes);

if ($maxSymbols > 0) {
    $symbols = array_slice($symbols, 0, $maxSymbols);
}

$cacheDir = __DIR__ . '/cache/candle_sync';
if (!is_dir($cacheDir)) {
    mkdir($cacheDir, 0755, true);
}

sync_ensure_candles_table();

echo "Candle Sync Build: " . SYNC_CANDLES_BUILD . PHP_EOL;
echo "Mode: {$mode}" . ($dryRun ? " (dry-run)" : "") . PHP_EOL;
echo "Symbols: " . count($symbols) . PHP_EOL;
echo "Timeframes: " . implode(', ', $timeframes) . PHP_EOL;
echo "Limit: {$outputLimit}" . PHP_EOL;
echo "Core only: " . ($coreOnly ? 'yes' : 'no') . PHP_EOL;
echo "Force: " . ($force ? 'yes' : 'no') . PHP_EOL;
echo "Asset filter: " . (!empty($assetTypeFilter) ? implode(',', $assetTypeFilter) : 'ALL') . PHP_EOL;
echo "Alert-aware mode: " . ($alertAware ? 'yes - active tracked setups only' : 'no') . PHP_EOL;
echo "Enterprise queue scheduler: " . ($enterpriseScheduler ? 'yes - queued/priority/budgeted' : 'no') . PHP_EOL;
if (is_array($queueBuildSummary)) {
    echo "Queue build: pairs_seen=" . (int)($queueBuildSummary['pairs_seen'] ?? 0)
        . "; jobs_upserted=" . (int)($queueBuildSummary['jobs_upserted'] ?? 0)
        . "; stale_marked=" . (int)($queueBuildSummary['stale_marked'] ?? 0)
        . PHP_EOL;
}
echo "Paid candle providers: " . ($allowPaidCandles ? 'allowed' : 'blocked by default') . PHP_EOL;
echo str_repeat('=', 70) . PHP_EOL;

if (!$symbols) {
    echo "No symbols selected. In alert-aware mode this means there are no active tracked setups. Otherwise add active symbols or pass --symbols=BTCUSD." . PHP_EOL;
    exit(0);
}

if (!$timeframes) {
    echo "No timeframes selected." . PHP_EOL;
    exit(0);
}

$totalInserted = 0;
$totalUpdated = 0;
$totalSkipped = 0;
$totalErrors = 0;
$processedPairs = 0;
$SYNC_LAST_PROVIDER_DEBUG = '';

foreach ($symbols as $symbolRow) {
    $symbol = strtoupper(trim((string)($symbolRow['symbol'] ?? $symbolRow['code'] ?? $symbolRow[0] ?? '')));

    if ($symbol === '') {
        continue;
    }

    echo PHP_EOL . "SYMBOL: {$symbol}" . PHP_EOL;
    echo str_repeat('-', 70) . PHP_EOL;

    $rowTimeframes = !empty($symbolRow['__sync_timeframes']) && is_array($symbolRow['__sync_timeframes'])
        ? $symbolRow['__sync_timeframes']
        : $timeframes;

    foreach ($rowTimeframes as $dbTimeframe) {
        if ($maxPairs > 0 && $processedPairs >= $maxPairs) {
            echo "Max pairs reached. Stopping." . PHP_EOL;
            break 2;
        }

        $dbTimeframe = sync_normalize_timeframe_code($dbTimeframe);
        $queueJobId = 0;
        if (!empty($symbolRow['__queue_job_ids']) && is_array($symbolRow['__queue_job_ids'])) {
            $queueJobId = (int)($symbolRow['__queue_job_ids'][$dbTimeframe] ?? 0);
        }
        $apiInterval = sync_api_interval($dbTimeframe);
        $assetType = sync_row_asset_type($symbolRow, $symbol);

        if (!empty($assetTypeFilter) && !in_array($assetType, $assetTypeFilter, true)) {
            $totalSkipped++;
            echo "[{$dbTimeframe}] skipped by asset filter ({$assetType})" . PHP_EOL;
            if ($queueJobId > 0) sync_queue_finish($queueJobId, 'skipped', '', 0, 0, 'Skipped by asset filter.', 900);
            continue;
        }

        /*
         * Production routing: crypto candles must use Binance only, even if
         * the symbols table or a CLI provider tries to route them elsewhere.
         */
        $provider = $assetType === 'CRYPTO'
            ? 'binance'
            : ($providerOverride ?: (!empty($symbolRow['__sync_alert_aware']) ? sync_alert_aware_default_provider($assetType) : sync_resolve_provider($symbolRow)));

        if ($mode === 'live' && $assetType !== 'CRYPTO' && !$allowPaidCandles) {
            $totalSkipped++;
            echo "[{$dbTimeframe}] skipped: {$assetType} candle sync is paid-provider protected. Use --allow-paid-candles=1 with an explicit provider if you really want this." . PHP_EOL;
            sync_log_run($symbol, $dbTimeframe, $provider, $outputLimit, 0, 0, 'guard_skipped', 'Paid candle sync blocked by production API guard.');
            if ($queueJobId > 0) sync_queue_finish($queueJobId, 'guard_skipped', $provider, 0, 0, 'Paid candle sync blocked by production API guard.', 1800);
            continue;
        }

        if ($mode === 'live' && sync_provider_in_cooldown($provider)) {
            if (sync_provider_has_available_api_rows($provider, 'candles')) {
                echo "[{$dbTimeframe}] legacy provider cooldown file exists for {$provider}, but an API-key row is still available; continuing with row-level rotation." . PHP_EOL;
            } else {
                $cooldown = sync_provider_cooldown_info($provider);
                $fallbackProvider = sync_alert_aware_fallback_provider($provider, $assetType, !empty($symbolRow['__sync_alert_aware']));

                if ($fallbackProvider !== '' && !sync_provider_in_cooldown($fallbackProvider)) {
                    echo "[{$dbTimeframe}] {$provider} cooldown active; using fallback provider={$fallbackProvider} for active setup." . PHP_EOL;
                    $provider = $fallbackProvider;
                    $apiInterval = sync_api_interval($dbTimeframe);
                } else {
                    $totalSkipped++;
                    $message = 'Provider cooldown active until ' . ($cooldown['until_label'] ?? 'unknown') . '. Reason: ' . ($cooldown['reason'] ?? 'provider guard');
                    echo "[{$dbTimeframe}] skipped: {$provider} cooldown active. " . $message . PHP_EOL;

                    if (sync_bool_option($options, 'log-cooldown', false) || sync_setting_enabled('candle_sync_log_provider_cooldown', false)) {
                        sync_log_run($symbol, $dbTimeframe, $provider, $outputLimit, 0, 0, 'provider_cooldown', $message);
                    }
                    if ($queueJobId > 0) sync_queue_finish($queueJobId, 'provider_cooldown', $provider, 0, 0, $message, 900);
                    continue;
                }
            }
        }

        $cacheKey = sync_candle_pair_cache_key($symbol, $dbTimeframe, $provider);
        $cacheFile = $cacheDir . '/' . $cacheKey . '.txt';
        $minSeconds = sync_minimum_seconds_between_calls($dbTimeframe, $mode);

        if (!$force && $mode === 'live' && is_file($cacheFile)) {
            $lastRun = (int)trim((string)@file_get_contents($cacheFile));
            $age = time() - $lastRun;

            if ($lastRun > 0 && $age < $minSeconds) {
                $totalSkipped++;
                echo "[{$dbTimeframe}] skipped by cache ({$age}s old, minimum {$minSeconds}s)" . PHP_EOL;
                if ($queueJobId > 0) sync_queue_finish($queueJobId, 'cache_skipped', $provider, 0, 0, "Pair cache still fresh ({$age}s old, minimum {$minSeconds}s).", max(30, $minSeconds - $age));
                continue;
            }
        }

        $processedPairs++;
        echo "[{$dbTimeframe}] provider={$provider}; api_interval={$apiInterval}; limit={$outputLimit}" . PHP_EOL;

        if ($dryRun) {
            echo "  dry-run: no API call, no database write." . PHP_EOL;
            continue;
        }

        try {
            $startedAt = microtime(true);
            $data = sync_fetch_candles($symbol, $apiInterval, $outputLimit, $provider);
            $elapsed = number_format(microtime(true) - $startedAt, 2);

            if (!$data || !is_array($data)) {
                echo "  no data returned ({$elapsed}s)." . PHP_EOL;
                $totalSkipped++;

                $providerDebug = (string)($GLOBALS['SYNC_LAST_PROVIDER_DEBUG'] ?? '');
                $status = 'no_data';
                $message = sync_short_log_message('No data returned', $providerDebug);

                if (sync_provider_limit_detected($provider, $providerDebug)) {
                    $status = 'provider_limit';
                    if (sync_provider_has_available_api_rows($provider, 'candles')) {
                        $message = sync_short_log_message('Provider limit detected on one API-key row. Row-level rotation/cooldown is active, so no provider-wide cooldown was set.', $providerDebug);
                        echo "  provider limit detected on one {$provider} API-key row; row-level rotation remains available." . PHP_EOL;
                    } else {
                        $cooldownMinutes = (int)sync_setting('candle_sync_' . strtolower($provider) . '_cooldown_minutes', sync_setting('candle_sync_paid_provider_cooldown_minutes', '180'));
                        $cooldownMinutes = max(15, min(1440, $cooldownMinutes));
                        sync_set_provider_cooldown($provider, $cooldownMinutes, sync_limit_reason($providerDebug));
                        $message = sync_short_log_message('Provider limit detected. Cooldown set for ' . $cooldownMinutes . ' minutes.', $providerDebug);
                        echo "  provider limit detected; {$provider} cooldown set for {$cooldownMinutes} minutes." . PHP_EOL;
                    }
                }

                sync_log_run($symbol, $dbTimeframe, $provider, $outputLimit, 0, 0, $status, $message);
                if ($queueJobId > 0) sync_queue_finish($queueJobId, $status, $provider, 0, 0, $message, max(300, min(1800, $minSeconds)));
                continue;
            }

            $inserted = 0;
            $updated = 0;
            $invalid = 0;
            $firstOpen = null;
            $lastOpen = null;

            foreach ($data as $rawCandle) {
                $candle = sync_normalize_candle($rawCandle);

                if (!$candle) {
                    $invalid++;
                    continue;
                }

                $openTime = $candle['datetime'];

                if (!sync_candle_open_allowed_for_symbol($symbol, $openTime, $dbTimeframe)) {
                    $invalid++;
                    continue;
                }

                $firstOpen = $firstOpen ?: $openTime;
                $lastOpen = $openTime;

                $write = sync_upsert_candle(
                    $symbol,
                    $dbTimeframe,
                    $openTime,
                    $candle['open'],
                    $candle['high'],
                    $candle['low'],
                    $candle['close'],
                    $candle['volume']
                );

                if ($write === 'inserted') {
                    $inserted++;
                } elseif ($write === 'updated') {
                    $updated++;
                }
            }

            file_put_contents($cacheFile, (string)time());

            $totalInserted += $inserted;
            $totalUpdated += $updated;

            echo "  inserted={$inserted}; updated={$updated}; invalid={$invalid}; returned=" . count($data) . "; elapsed={$elapsed}s" . PHP_EOL;
            if ($firstOpen || $lastOpen) {
                echo "  range={$firstOpen} -> {$lastOpen}" . PHP_EOL;
            }

            $okMessage = "returned=" . count($data) . "; invalid={$invalid}";
            sync_log_run($symbol, $dbTimeframe, $provider, $outputLimit, $inserted, $updated, 'ok', $okMessage);
            if ($queueJobId > 0) sync_queue_finish($queueJobId, 'ok', $provider, $inserted, $updated, $okMessage, $minSeconds);

            usleep(sync_api_delay_microseconds($provider));
        } catch (Throwable $e) {
            $totalErrors++;
            echo "  ERROR: " . $e->getMessage() . PHP_EOL;
            $errorMessage = sync_short_log_message($e->getMessage(), $GLOBALS['SYNC_LAST_PROVIDER_DEBUG'] ?? '');
            sync_log_run($symbol, $dbTimeframe, $provider, $outputLimit, 0, 0, 'error', $errorMessage);
            if ($queueJobId > 0) sync_queue_finish($queueJobId, 'error', $provider, 0, 0, $errorMessage, 900);
            usleep(500000);
        }
    }
}

echo PHP_EOL . str_repeat('=', 70) . PHP_EOL;
echo "DONE" . PHP_EOL;
echo "Pairs processed: {$processedPairs}" . PHP_EOL;
echo "Inserted: {$totalInserted}" . PHP_EOL;
echo "Updated: {$totalUpdated}" . PHP_EOL;
echo "Skipped: {$totalSkipped}" . PHP_EOL;
echo "Errors: {$totalErrors}" . PHP_EOL;
echo "UTC: " . gmdate('Y-m-d H:i:s') . PHP_EOL;

function sync_cli_options(array $argv): array
{
    $options = [];

    for ($i = 1; $i < count($argv); $i++) {
        $arg = (string)$argv[$i];

        if (strpos($arg, '--') !== 0) {
            continue;
        }

        $arg = substr($arg, 2);

        if (strpos($arg, '=') !== false) {
            [$key, $value] = explode('=', $arg, 2);
            $options[$key] = $value;
        } else {
            $key = $arg;
            $next = $argv[$i + 1] ?? null;

            if ($next !== null && strpos((string)$next, '--') !== 0) {
                $options[$key] = $next;
                $i++;
            } else {
                $options[$key] = '1';
            }
        }
    }

    return $options;
}

function sync_bool_option(array $options, string $key, bool $default = false): bool
{
    if (!array_key_exists($key, $options)) {
        return $default;
    }

    $value = strtolower(trim((string)$options[$key]));

    return in_array($value, ['1', 'true', 'yes', 'on'], true);
}

function sync_setting(string $key, string $default = ''): string
{
    try {
        $stmt = db()->prepare("SELECT setting_value FROM system_settings WHERE setting_key = ? LIMIT 1");
        $stmt->execute([$key]);
        $value = $stmt->fetchColumn();

        return $value !== false ? (string)$value : $default;
    } catch (Throwable $e) {
        return $default;
    }
}

function sync_setting_enabled(string $key, bool $default = false): bool
{
    $value = strtolower(trim(sync_setting($key, $default ? '1' : '0')));
    if ($value === '') {
        return $default;
    }

    return in_array($value, ['1', 'true', 'yes', 'on', 'enabled'], true);
}

function sync_requested_timeframes(array $options): array
{
    $requested = trim((string)($options['timeframes'] ?? $options['tf'] ?? ''));

    if ($requested === '') {
        $requested = sync_setting('candle_sync_timeframes', '1M,5M,15M,30M,1H,4H,1D');
    }

    $items = array_values(array_filter(array_map('trim', explode(',', $requested))));

    $normalized = [];
    foreach ($items as $item) {
        $code = sync_normalize_timeframe_code($item);
        if ($code !== '' && !in_array($code, $normalized, true)) {
            $normalized[] = $code;
        }
    }

    return $normalized;
}


function sync_requested_asset_types(array $options): array
{
    $raw = trim((string)($options['asset-types'] ?? $options['asset_type'] ?? $options['asset-types-filter'] ?? ''));
    if ($raw === '') return [];
    $items = array_values(array_filter(array_map('trim', explode(',', strtoupper($raw)))));
    $valid = ['CRYPTO','FOREX','METAL','METALS','INDEX','INDICES','OTHER'];
    $out = [];
    foreach ($items as $item) {
        if ($item === 'METALS') $item = 'METAL';
        if ($item === 'INDEX') $item = 'INDICES';
        if (in_array($item, $valid, true) && !in_array($item, $out, true)) $out[] = $item;
    }
    return $out;
}

function sync_row_asset_type(array $symbolRow, string $symbol): string
{
    $assetType = strtoupper(trim((string)($symbolRow['asset_type'] ?? $symbolRow['type'] ?? '')));
    if ($assetType === 'METALS') $assetType = 'METAL';
    if ($assetType === 'INDEX') $assetType = 'INDICES';
    if ($assetType !== '') return $assetType;
    return sync_guess_asset_type($symbol);
}


function sync_active_alert_symbol_rows(array $options, array $fallbackTimeframes): array
{
    $requestedSymbols = sync_cli_symbol_filter($options);
    $requestedAssets = sync_requested_asset_types($options);

    /*
     * Important scaling rule:
     * Do not read N alert rows and then sync once per row. Instead, ask MySQL
     * for distinct active symbol/setup-timeframe pairs, then merge the mapped
     * confirmation timeframe in PHP. This lets 500 active setups on the same
     * EURUSD 1H/15M flow become only EURUSD 1H + EURUSD 15M candle requests.
     */
    $maxDistinctPairs = isset($options['max-pairs'])
        ? max(1, min(5000, (int)$options['max-pairs']))
        : (int)sync_setting('candle_sync_alert_aware_max_distinct_pairs', sync_setting('candle_sync_alert_aware_max_alerts', '500'));
    $maxDistinctPairs = max(1, min(5000, $maxDistinctPairs));

    /*
     * Credit-safe default: do not spend candle credits on early tracking stages.
     * Stage 1/2 detection should use live quote first; closed-candle syncing is
     * mainly required after liquidity sweep when waiting for BOS/CHoCH.
     * Set candle_sync_alert_aware_stage_filter to "ALL" or empty to include all.
     */
    $stageFilterRaw = strtoupper(trim(sync_setting('candle_sync_alert_aware_stage_filter', 'LIQUIDITY_SWEPT,STRUCTURE_CONFIRMED')));
    $stageFilterAll = ($stageFilterRaw === '' || $stageFilterRaw === 'ALL' || $stageFilterRaw === '*');
    $stageFilter = $stageFilterAll ? [] : array_values(array_filter(array_map('trim', explode(',', $stageFilterRaw))));

    $symbolExpr = "UPPER(REPLACE(REPLACE(REPLACE(REPLACE(symbol, '/', ''), '-', ''), '_', ''), ' ', ''))";
    $where = [
        "is_active = 1",
        "triggered_at IS NULL",
        "(expired_at IS NULL OR expired_at = '0000-00-00 00:00:00')",
        "COALESCE(lifecycle_stage, 'TRACKING') NOT IN ('ALERT_SENT', 'INVALIDATED', 'EXPIRED')",
        "symbol IS NOT NULL",
        "symbol <> ''",
        "timeframe IS NOT NULL",
        "timeframe <> ''",
    ];
    $params = [];

    if (!$stageFilterAll && !empty($stageFilter)) {
        $where[] = "UPPER(COALESCE(lifecycle_stage, 'TRACKING')) IN (" . implode(',', array_fill(0, count($stageFilter), '?')) . ")";
        foreach ($stageFilter as $stage) {
            $params[] = $stage;
        }
    }

    if (!empty($requestedSymbols)) {
        $where[] = $symbolExpr . " IN (" . implode(',', array_fill(0, count($requestedSymbols), '?')) . ")";
        foreach ($requestedSymbols as $symbol) {
            $params[] = $symbol;
        }
    }

    try {
        $sql = "
            SELECT
                {$symbolExpr} AS symbol,
                timeframe,
                COUNT(*) AS alert_count,
                MIN(COALESCE(last_checked_at, tracking_started_at, created_at)) AS sort_time
            FROM institutional_alerts
            WHERE " . implode(' AND ', $where) . "
            GROUP BY {$symbolExpr}, timeframe
            ORDER BY sort_time ASC, symbol ASC, timeframe ASC
            LIMIT {$maxDistinctPairs}
        ";
        $stmt = db()->prepare($sql);
        $stmt->execute($params);
        $pairs = $stmt->fetchAll(PDO::FETCH_ASSOC) ?: [];
    } catch (Throwable $e) {
        echo "Alert-aware distinct-pair lookup failed: " . $e->getMessage() . PHP_EOL;
        return sync_requested_symbols($options, true, false);
    }

    $rows = [];

    foreach ($pairs as $pair) {
        $symbol = strtoupper(trim((string)($pair['symbol'] ?? '')));
        if ($symbol === '') {
            continue;
        }

        $assetType = sync_guess_asset_type($symbol);
        if ($requestedAssets && !in_array($assetType, $requestedAssets, true)) {
            continue;
        }

        $setupTf = sync_normalize_timeframe_code((string)($pair['timeframe'] ?? ''));
        if ($setupTf === '') {
            continue;
        }

        $confirmTf = sync_confirmation_timeframe_for_setup($setupTf);
        $needed = [$setupTf];

        if ($confirmTf !== '' && !in_array($confirmTf, $needed, true)) {
            $needed[] = $confirmTf;
        }

        /*
         * Optional fallback protection:
         * If a caller supplied fallback timeframes and wants to force an extra TF
         * for testing, allow --include-requested-timeframes=1.
         */
        if (sync_bool_option($options, 'include-requested-timeframes', false)) {
            foreach ($fallbackTimeframes as $tf) {
                $tf = sync_normalize_timeframe_code($tf);
                if ($tf !== '' && !in_array($tf, $needed, true)) {
                    $needed[] = $tf;
                }
            }
        }

        if (!isset($rows[$symbol])) {
            $rows[$symbol] = [
                'symbol' => $symbol,
                'asset_type' => $assetType,
                'provider' => $assetType === 'CRYPTO' ? 'binance' : 'fmp',
                '__sync_timeframes' => [],
                '__sync_alert_aware' => true,
                '__sync_alert_count' => 0,
            ];
        }

        $rows[$symbol]['__sync_alert_count'] += (int)($pair['alert_count'] ?? 0);

        foreach ($needed as $tf) {
            if (!in_array($tf, $rows[$symbol]['__sync_timeframes'], true)) {
                $rows[$symbol]['__sync_timeframes'][] = $tf;
            }
        }
    }

    foreach ($rows as &$row) {
        usort($row['__sync_timeframes'], 'sync_timeframe_sort_compare');
    }
    unset($row);

    return array_values($rows);
}

function sync_deduplicate_symbol_rows(array $rows, array $fallbackTimeframes = []): array
{
    $deduped = [];

    foreach ($rows as $row) {
        if (!is_array($row)) {
            $symbol = strtoupper(str_replace(['/', '-', '_', ' '], '', trim((string)$row)));
            $row = ['symbol' => $symbol, 'asset_type' => sync_guess_asset_type($symbol)];
        }

        $symbol = strtoupper(str_replace(['/', '-', '_', ' '], '', trim((string)($row['symbol'] ?? $row['code'] ?? $row[0] ?? ''))));
        if ($symbol === '') {
            continue;
        }

        $assetType = sync_row_asset_type($row, $symbol);
        $timeframes = !empty($row['__sync_timeframes']) && is_array($row['__sync_timeframes'])
            ? $row['__sync_timeframes']
            : $fallbackTimeframes;

        if (!isset($deduped[$symbol])) {
            $row['symbol'] = $symbol;
            $row['asset_type'] = $assetType;
            $row['__sync_timeframes'] = [];
            $deduped[$symbol] = $row;
        }

        foreach ($timeframes as $tf) {
            $tf = sync_normalize_timeframe_code($tf);
            if ($tf !== '' && !in_array($tf, $deduped[$symbol]['__sync_timeframes'], true)) {
                $deduped[$symbol]['__sync_timeframes'][] = $tf;
            }
        }

        if (isset($row['__sync_alert_count'])) {
            $deduped[$symbol]['__sync_alert_count'] = (int)($deduped[$symbol]['__sync_alert_count'] ?? 0) + (int)$row['__sync_alert_count'];
        }
    }

    foreach ($deduped as &$row) {
        usort($row['__sync_timeframes'], 'sync_timeframe_sort_compare');
    }
    unset($row);

    return array_values($deduped);
}

function sync_candle_pair_cache_key(string $symbol, string $timeframe, string $provider = ''): string
{
    $symbol = strtoupper(str_replace(['/', '-', '_', ' '], '', trim($symbol)));
    $timeframe = sync_normalize_timeframe_code($timeframe);

    /*
     * Default: cache candle sync by symbol+timeframe, not by provider.
     * This prevents FMP failure/fallback to TwelveData from causing an immediate
     * second paid candle request for the same pair in the same cache window.
     */
    $providerIndependent = sync_setting_enabled('candle_sync_pair_cache_provider_independent', true);
    $prefix = $providerIndependent ? 'pair' : strtolower(trim($provider));
    if ($prefix === '') {
        $prefix = 'pair';
    }

    return preg_replace('/[^A-Za-z0-9_\-]/', '_', $prefix . '_' . $symbol . '_' . $timeframe);
}

function sync_cli_symbol_filter(array $options): array
{
    $requested = trim((string)($options['symbols'] ?? $options['symbol'] ?? ''));
    if ($requested === '') {
        return [];
    }

    $out = [];
    foreach (array_values(array_filter(array_map('trim', explode(',', $requested)))) as $symbol) {
        $symbol = strtoupper(str_replace(['/', '-', '_', ' '], '', $symbol));
        if ($symbol !== '' && !in_array($symbol, $out, true)) {
            $out[] = $symbol;
        }
    }

    return $out;
}

function sync_confirmation_timeframe_for_setup(string $setupTimeframe): string
{
    $setup = sync_normalize_timeframe_code($setupTimeframe);
    $configured = trim(sync_setting('alert_confirmation_timeframe_map', ''));

    if ($configured !== '') {
        $json = json_decode($configured, true);
        if (is_array($json)) {
            foreach ($json as $from => $to) {
                if (sync_normalize_timeframe_code($from) === $setup) {
                    return sync_normalize_timeframe_code($to);
                }
            }
        } else {
            foreach (explode(',', $configured) as $pair) {
                $parts = array_map('trim', explode(':', $pair, 2));
                if (count($parts) === 2 && sync_normalize_timeframe_code($parts[0]) === $setup) {
                    return sync_normalize_timeframe_code($parts[1]);
                }
            }
        }
    }

    $map = [
        '1M' => '1M',
        '3M' => '1M',
        '5M' => '1M',
        '15M' => '5M',
        '30M' => '5M',
        '1H' => '15M',
        '2H' => '15M',
        '4H' => '1H',
        '1D' => '4H',
        '1W' => '1D',
    ];

    return $map[$setup] ?? $setup;
}

function sync_timeframe_sort_compare($a, $b): int
{
    return sync_timeframe_seconds($a) <=> sync_timeframe_seconds($b);
}

function sync_timeframe_seconds($timeframe): int
{
    $tf = sync_normalize_timeframe_code($timeframe);
    $map = [
        '1M' => 60, '3M' => 180, '5M' => 300, '15M' => 900, '30M' => 1800,
        '1H' => 3600, '2H' => 7200, '4H' => 14400, '1D' => 86400, '1W' => 604800,
    ];

    return $map[$tf] ?? 999999;
}


function sync_requested_symbols(array $options, bool $coreOnly, bool $includeInactive): array
{
    $requested = trim((string)($options['symbols'] ?? $options['symbol'] ?? ''));

    if ($requested !== '') {
        $rows = [];
        foreach (array_values(array_filter(array_map('trim', explode(',', $requested)))) as $symbol) {
            $rows[] = [
                'symbol' => strtoupper($symbol),
                'provider' => strtolower((string)($options['provider'] ?? '')),
                'asset_type' => sync_guess_asset_type($symbol),
            ];
        }

        return $rows;
    }

    try {
        if (class_exists('SystemConfigService')) {
            $rows = SystemConfigService::getActiveSymbols($coreOnly);
            if (is_array($rows) && $rows) {
                return $rows;
            }
        }
    } catch (Throwable $e) {
    }

    try {
        $where = $includeInactive ? "1=1" : "COALESCE(is_active,1)=1";
        $stmt = db()->query("SELECT * FROM symbols WHERE {$where} ORDER BY COALESCE(sort_order, 999999), symbol ASC");
        return $stmt->fetchAll(PDO::FETCH_ASSOC) ?: [];
    } catch (Throwable $e) {
        return [];
    }
}

function sync_normalize_timeframe_code($timeframe): string
{
    $tf = strtoupper(str_replace([' ', '_'], '', trim((string)$timeframe)));

    $map = [
        'M1' => '1M', '1MIN' => '1M', '1MINUTE' => '1M', '1M' => '1M',
        'M3' => '3M', '3MIN' => '3M', '3MINUTE' => '3M', '3M' => '3M',
        'M5' => '5M', '5MIN' => '5M', '5MINUTE' => '5M', '5M' => '5M',
        'M15' => '15M', '15MIN' => '15M', '15MINUTE' => '15M', '15M' => '15M',
        'M30' => '30M', '30MIN' => '30M', '30MINUTE' => '30M', '30M' => '30M',
        'H1' => '1H', '1HOUR' => '1H', '1H' => '1H',
        'H2' => '2H', '2HOUR' => '2H', '2H' => '2H',
        'H4' => '4H', '4HOUR' => '4H', '4H' => '4H',
        'D1' => '1D', '1DAY' => '1D', '1D' => '1D',
        'W1' => '1W', '1WEEK' => '1W', '1W' => '1W',
    ];

    return $map[$tf] ?? $tf;
}

function sync_api_interval(string $timeframe): string
{
    $timeframe = sync_normalize_timeframe_code($timeframe);

    $map = [
        '1M' => '1min',
        '3M' => '3min',
        '5M' => '5min',
        '15M' => '15min',
        '30M' => '30min',
        '1H' => '1h',
        '2H' => '2h',
        '4H' => '4h',
        '1D' => '1day',
        '1W' => '1week',
    ];

    try {
        if (class_exists('SystemConfigService') && method_exists('SystemConfigService', 'getApiInterval')) {
            $fromDb = SystemConfigService::getApiInterval($timeframe);
            if ($fromDb) {
                return (string)$fromDb;
            }
        }
    } catch (Throwable $e) {
    }

    return $map[$timeframe] ?? strtolower($timeframe);
}

function sync_resolve_provider(array $symbolRow): string
{
    $assetType = sync_row_asset_type($symbolRow, (string)($symbolRow['symbol'] ?? $symbolRow['code'] ?? ''));

    /* Crypto candle routing is hard-forced to Binance to avoid TwelveData credit use. */
    if ($assetType === 'CRYPTO') {
        return 'binance';
    }

    foreach (['provider', 'data_provider', 'candle_provider', 'market_data_provider', 'api_provider'] as $key) {
        if (!empty($symbolRow[$key])) {
            return strtolower(trim((string)$symbolRow[$key]));
        }
    }

    return 'auto';
}

function sync_guess_asset_type(string $symbol): string
{
    $s = strtoupper($symbol);

    if (preg_match('/^(BTC|ETH|SOL|BNB|XRP|ADA|DOGE|AVAX|LINK|LTC|BCH|DOT|MATIC)/', $s)) {
        return 'CRYPTO';
    }

    if (in_array($s, ['XAUUSD', 'XAGUSD', 'XPTUSD', 'XPDUSD'], true)) {
        return 'METAL';
    }

    if (preg_match('/^[A-Z]{6}$/', $s)) {
        return 'FOREX';
    }

    if (preg_match('/^(NAS|US30|DJI|SPX|US100|US500|GER40|UK100|JP225)/', $s)) {
        return 'INDICES';
    }

    return 'OTHER';
}

function sync_fetch_candles(string $symbol, string $apiInterval, int $limit, string $provider): array
{
    $GLOBALS['SYNC_LAST_PROVIDER_DEBUG'] = '';
    if (!class_exists('MarketDataService')) {
        throw new RuntimeException('MarketDataService.php not found or class not loaded.');
    }

    $provider = strtolower(trim($provider)) ?: 'auto';

    /*
     * Use reflection so this cron remains compatible with older MarketDataService
     * versions that accept only 3 parameters and newer versions that accept provider.
     */
    $method = new ReflectionMethod('MarketDataService', 'fetchCandles');
    $paramCount = $method->getNumberOfParameters();

    ob_start();

    try {
        if ($paramCount >= 4) {
            $data = MarketDataService::fetchCandles($symbol, $apiInterval, $limit, $provider);
        } else {
            $data = MarketDataService::fetchCandles($symbol, $apiInterval, $limit);
        }
    } finally {
        $providerOutput = trim((string)ob_get_clean());
        $GLOBALS['SYNC_LAST_PROVIDER_DEBUG'] = $providerOutput;
        if ($providerOutput !== '') {
            foreach (explode("\n", $providerOutput) as $line) {
                $line = trim($line);
                if ($line !== '') {
                    echo "  provider: {$line}" . PHP_EOL;
                }
            }
        }
    }

    return is_array($data ?? null) ? $data : [];
}

function sync_normalize_candle($raw): ?array
{
    if (!is_array($raw)) {
        return null;
    }

    $openTime = $raw['datetime']
        ?? $raw['date']
        ?? $raw['time']
        ?? $raw['timestamp']
        ?? $raw['open_time']
        ?? null;

    $open = $raw['open'] ?? $raw['open_price'] ?? null;
    $high = $raw['high'] ?? $raw['high_price'] ?? null;
    $low = $raw['low'] ?? $raw['low_price'] ?? null;
    $close = $raw['close'] ?? $raw['close_price'] ?? null;
    $volume = $raw['volume'] ?? null;

    if ($openTime === null || $open === null || $high === null || $low === null || $close === null) {
        return null;
    }

    $openTime = sync_normalize_open_time($openTime);

    if (!$openTime) {
        return null;
    }

    if (!is_numeric($open) || !is_numeric($high) || !is_numeric($low) || !is_numeric($close)) {
        return null;
    }

    return [
        'datetime' => $openTime,
        'open' => (float)$open,
        'high' => (float)$high,
        'low' => (float)$low,
        'close' => (float)$close,
        'volume' => is_numeric($volume) ? (float)$volume : null,
    ];
}

function sync_normalize_open_time($value): ?string
{
    if ($value === null || $value === '') {
        return null;
    }

    try {
        if (is_numeric($value)) {
            $ts = (int)$value;
            if ($ts > 20000000000) {
                $ts = (int)floor($ts / 1000);
            }

            return gmdate('Y-m-d H:i:s', $ts);
        }

        $value = trim((string)$value);

        if (preg_match('/^\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}$/', $value)) {
            /*
             * Provider-normalized database times are treated as UTC.
             */
            return $value;
        }

        $dt = new DateTimeImmutable($value, new DateTimeZone('UTC'));

        return $dt->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s');
    } catch (Throwable $e) {
        return null;
    }
}


function sync_candle_open_allowed_for_symbol(string $symbol, string $openTime, string $timeframe = ''): bool
{
    if (!sync_setting_enabled('prevent_weekend_candle_ingestion', true)) {
        return true;
    }

    $assetType = sync_asset_type_for_symbol($symbol);
    if ($assetType === 'CRYPTO') {
        return true;
    }

    $ts = strtotime($openTime . ' UTC');
    if (!$ts) {
        return false;
    }

    return sync_market_open_at_timestamp($symbol, $ts);
}

function sync_market_open_at_timestamp(string $symbol, int $timestamp): bool
{
    $assetType = sync_asset_type_for_symbol($symbol);
    if ($assetType === 'CRYPTO') {
        return true;
    }

    try {
        $tz = new DateTimeZone(sync_setting('market_open_guard_timezone', 'America/New_York'));
    } catch (Throwable $e) {
        $tz = new DateTimeZone('America/New_York');
    }

    $dt = (new DateTimeImmutable('@' . $timestamp))->setTimezone($tz);
    $dow = (int)$dt->format('N');
    $minutes = ((int)$dt->format('G') * 60) + (int)$dt->format('i');
    $closeMinutes = 17 * 60;

    if ($dow === 6) {
        return false;
    }
    if ($dow === 5 && $minutes >= $closeMinutes) {
        return false;
    }
    if ($dow === 7 && $minutes < $closeMinutes) {
        return false;
    }

    return true;
}

function sync_asset_type_for_symbol(string $symbol): string
{
    $s = strtoupper(str_replace(['/', '-', '_', ' '], '', trim($symbol)));

    if (preg_match('/^(BTC|ETH|SOL|BNB|XRP|ADA|DOGE|AVAX|LINK|LTC|BCH|DOT|MATIC|TRX|ATOM|NEAR|APT|OP|ARB)/', $s)) {
        return 'CRYPTO';
    }
    if (in_array($s, ['XAUUSD', 'XAGUSD', 'XPTUSD', 'XPDUSD'], true)) {
        return 'METAL';
    }
    if (preg_match('/^[A-Z]{6}$/', $s)) {
        return 'FOREX';
    }
    if (preg_match('/^(NAS|US30|DJI|SPX|US100|US500|GER40|UK100|JP225)/', $s)) {
        return 'INDICES';
    }

    return 'OTHER';
}

function sync_upsert_candle(string $symbol, string $timeframe, string $openTime, float $open, float $high, float $low, float $close, $volume = null): string
{
    $stmt = db()->prepare("
        SELECT id
        FROM candles
        WHERE symbol = ?
          AND timeframe = ?
          AND open_time = ?
        LIMIT 1
    ");
    $stmt->execute([$symbol, $timeframe, $openTime]);
    $id = $stmt->fetchColumn();

    if ($id) {
        $stmt = db()->prepare("
            UPDATE candles
            SET open_price = ?,
                high_price = ?,
                low_price = ?,
                close_price = ?,
                volume = ?
            WHERE symbol = ?
              AND timeframe = ?
              AND open_time = ?
        ");
        $stmt->execute([$open, $high, $low, $close, $volume, $symbol, $timeframe, $openTime]);

        return 'updated';
    }

    $stmt = db()->prepare("
        INSERT INTO candles
            (symbol, timeframe, open_time, open_price, high_price, low_price, close_price, volume)
        VALUES
            (?, ?, ?, ?, ?, ?, ?, ?)
    ");
    $stmt->execute([$symbol, $timeframe, $openTime, $open, $high, $low, $close, $volume]);

    return 'inserted';
}

function sync_ensure_candles_table(): void
{
    try {
        db()->exec("
            CREATE TABLE IF NOT EXISTS candles (
                id INT AUTO_INCREMENT PRIMARY KEY,
                symbol VARCHAR(30) NOT NULL,
                timeframe VARCHAR(20) NOT NULL,
                open_time DATETIME NOT NULL,
                open_price DECIMAL(20,8) NOT NULL,
                high_price DECIMAL(20,8) NOT NULL,
                low_price DECIMAL(20,8) NOT NULL,
                close_price DECIMAL(20,8) NOT NULL,
                volume DECIMAL(24,8) NULL,
                UNIQUE KEY uniq_symbol_timeframe_open (symbol, timeframe, open_time),
                INDEX idx_symbol_timeframe_open (symbol, timeframe, open_time)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        ");
    } catch (Throwable $e) {
        /*
         * Existing tables with a different schema are allowed. The upsert path
         * below uses SELECT then UPDATE/INSERT, so it does not depend on a new
         * unique key being created here.
         */
    }
}


function sync_short_log_message(string $message, string $debug = ''): string
{
    $message = trim($message);
    $debug = trim($debug);

    if ($debug !== '') {
        $debug = preg_replace('/\s+/', ' ', $debug);
        $message .= ' | Provider debug: ' . $debug;
    }

    if (strlen($message) > 1800) {
        $message = substr($message, 0, 1800) . '...';
    }

    return $message;
}

function sync_log_run(string $symbol, string $timeframe, string $provider, int $limit, int $inserted, int $updated, string $status, string $message): void
{
    try {
        db()->exec("
            CREATE TABLE IF NOT EXISTS candle_sync_logs (
                id INT AUTO_INCREMENT PRIMARY KEY,
                symbol VARCHAR(30) NOT NULL,
                timeframe VARCHAR(20) NOT NULL,
                provider VARCHAR(40) NULL,
                request_limit INT NULL,
                inserted_count INT DEFAULT 0,
                updated_count INT DEFAULT 0,
                status VARCHAR(30) NOT NULL,
                message TEXT NULL,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                INDEX idx_symbol_timeframe_created (symbol, timeframe, created_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        ");

        $stmt = db()->prepare("
            INSERT INTO candle_sync_logs
                (symbol, timeframe, provider, request_limit, inserted_count, updated_count, status, message)
            VALUES
                (?, ?, ?, ?, ?, ?, ?, ?)
        ");
        $stmt->execute([$symbol, $timeframe, $provider, $limit, $inserted, $updated, $status, $message]);
    } catch (Throwable $e) {
        /*
         * Logging should never stop candle syncing.
         */
    }
}


function sync_alert_aware_default_provider(string $assetType): string
{
    $assetType = strtoupper(trim($assetType));

    if ($assetType === 'METAL') {
        $provider = strtolower(trim(sync_setting('candle_sync_alert_aware_metal_provider', 'twelvedata')));
        if ($provider === '' || $provider === 'auto') {
            return 'twelvedata';
        }
        if ($provider === 'fmp' && !sync_setting_enabled('candle_sync_fmp_metals_enabled', false)) {
            return 'twelvedata';
        }
        return $provider;
    }

    if ($assetType === 'FOREX') {
        return strtolower(trim(sync_setting('candle_sync_alert_aware_forex_provider', 'fmp'))) ?: 'fmp';
    }

    return strtolower(trim(sync_setting('candle_sync_alert_aware_default_provider', 'fmp'))) ?: 'fmp';
}

function sync_alert_aware_fallback_provider(string $provider, string $assetType, bool $alertAware): string
{
    if (!$alertAware || $assetType === 'CRYPTO') {
        return '';
    }

    $fallback = strtolower(trim(sync_setting('candle_sync_alert_aware_noncrypto_fallback_provider', 'twelvedata')));
    if ($fallback === '' || $fallback === 'none' || $fallback === strtolower(trim($provider))) {
        return '';
    }

    return $fallback;
}

function sync_provider_has_available_api_rows(string $provider, string $capability = 'candles'): bool
{
    $provider = strtolower(trim($provider));
    if ($provider === '' || $provider === 'binance' || $provider === 'auto' || $provider === 'blocked') {
        return false;
    }

    if (class_exists('ApiCreditGuard') && method_exists('ApiCreditGuard', 'providerRows')) {
        try {
            return (bool)ApiCreditGuard::providerRows($provider, $capability, 1, true);
        } catch (Throwable $e) {
            return false;
        }
    }

    if (class_exists('ApiProviderRotationService') && method_exists('ApiProviderRotationService', 'getBestProvider')) {
        try {
            return (bool)ApiProviderRotationService::getBestProvider($capability, null, null, $provider);
        } catch (Throwable $e) {
            return false;
        }
    }

    return false;
}

function sync_provider_cooldown_file(string $provider): string
{
    $provider = preg_replace('/[^a-z0-9_\-]/', '_', strtolower(trim($provider)));
    $dir = __DIR__ . '/cache/candle_sync';
    if (!is_dir($dir)) {
        @mkdir($dir, 0755, true);
    }

    return $dir . '/provider_' . $provider . '_cooldown.json';
}

function sync_provider_cooldown_info(string $provider): array
{
    $file = sync_provider_cooldown_file($provider);
    if (!is_file($file)) {
        return [];
    }

    $data = json_decode((string)@file_get_contents($file), true);
    if (!is_array($data)) {
        return [];
    }

    $until = (int)($data['until'] ?? 0);
    $data['until_label'] = $until > 0 ? gmdate('Y-m-d H:i:s', $until) . ' UTC' : '';

    return $data;
}

function sync_provider_in_cooldown(string $provider): bool
{
    $info = sync_provider_cooldown_info($provider);
    $until = (int)($info['until'] ?? 0);

    if ($until <= 0) {
        return false;
    }

    if ($until <= time()) {
        @unlink(sync_provider_cooldown_file($provider));
        return false;
    }

    return true;
}

function sync_set_provider_cooldown(string $provider, int $minutes, string $reason): void
{
    $until = time() + (max(1, $minutes) * 60);
    $payload = [
        'provider' => strtolower(trim($provider)),
        'until' => $until,
        'until_utc' => gmdate('Y-m-d H:i:s', $until),
        'reason' => trim($reason) ?: 'provider limit detected',
        'set_at' => gmdate('Y-m-d H:i:s'),
    ];

    @file_put_contents(sync_provider_cooldown_file($provider), json_encode($payload, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES));
}

function sync_provider_limit_detected(string $provider, string $debug): bool
{
    $provider = strtolower(trim($provider));
    $debug = strtolower($debug);

    if ($debug === '') {
        return false;
    }

    if (strpos($debug, 'limit reach') !== false) {
        return true;
    }

    if (strpos($debug, 'rate limit') !== false || strpos($debug, 'daily limit') !== false || strpos($debug, 'api limit') !== false) {
        return true;
    }

    if (strpos($debug, 'credits') !== false || strpos($debug, 'api credits') !== false || strpos($debug, 'quota') !== false) {
        return true;
    }

    if ($provider === 'fmp' && (strpos($debug, 'upgrade your plan') !== false || strpos($debug, 'please upgrade') !== false)) {
        return true;
    }

    return false;
}

function sync_limit_reason(string $debug): string
{
    $debug = trim(preg_replace('/\s+/', ' ', $debug));
    if ($debug === '') {
        return 'provider limit detected';
    }

    if (preg_match('/Limit Reach[^\\n\\r]*/i', $debug, $m)) {
        return trim($m[0]);
    }

    if (preg_match('/[^.]*credits[^.]*\./i', $debug, $m)) {
        return trim($m[0]);
    }

    if (preg_match('/rate limit[^\\n\\r]*/i', $debug, $m)) {
        return trim($m[0]);
    }

    return substr($debug, 0, 240);
}


function sync_minimum_seconds_between_calls(string $timeframe, string $mode): int
{
    if ($mode === 'backfill') {
        return 0;
    }

    $timeframe = sync_normalize_timeframe_code($timeframe);

    $defaults = [
        // Credit-safe defaults. These still keep active tracked setups fresh,
        // but avoid re-calling paid candle providers every minute.
        '1M' => 60,
        '3M' => 120,
        '5M' => 240,
        '15M' => 600,
        '30M' => 900,
        '1H' => 1800,
        '2H' => 2700,
        '4H' => 3600,
        '1D' => 14400,
        '1W' => 43200,
    ];

    $settingKey = 'candle_sync_min_seconds_' . strtolower($timeframe);
    $fromSetting = sync_setting($settingKey, '');

    if ($fromSetting !== '' && is_numeric($fromSetting)) {
        return max(0, (int)$fromSetting);
    }

    return $defaults[$timeframe] ?? 300;
}


function sync_table_exists(string $table): bool
{
    static $cache = [];
    $table = preg_replace('/[^A-Za-z0-9_]/', '', $table);
    if ($table === '') return false;
    if (array_key_exists($table, $cache)) return $cache[$table];
    try {
        $stmt = db()->prepare("SHOW TABLES LIKE ?");
        $stmt->execute([$table]);
        return $cache[$table] = (bool)$stmt->fetchColumn();
    } catch (Throwable $e) {
        return $cache[$table] = false;
    }
}

function sync_column_exists(string $table, string $column): bool
{
    static $cache = [];
    $table = preg_replace('/[^A-Za-z0-9_]/', '', $table);
    $column = preg_replace('/[^A-Za-z0-9_]/', '', $column);
    if ($table === '' || $column === '') return false;
    $key = $table . '.' . $column;
    if (array_key_exists($key, $cache)) return $cache[$key];
    try {
        $stmt = db()->prepare("SHOW COLUMNS FROM `$table` LIKE ?");
        $stmt->execute([$column]);
        return $cache[$key] = (bool)$stmt->fetchColumn();
    } catch (Throwable $e) {
        return $cache[$key] = false;
    }
}

function sync_ensure_candle_queue_tables(): void
{
    try {
        db()->exec("
            CREATE TABLE IF NOT EXISTS candle_sync_queue (
                id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
                symbol VARCHAR(40) NOT NULL,
                timeframe VARCHAR(20) NOT NULL,
                asset_type VARCHAR(30) NULL,
                provider VARCHAR(60) NULL,
                priority_score DECIMAL(12,4) DEFAULT 0,
                priority_bucket VARCHAR(40) NULL,
                alert_count INT DEFAULT 0,
                free_count INT DEFAULT 0,
                paid_count INT DEFAULT 0,
                stage_rank INT DEFAULT 0,
                min_interval_seconds INT DEFAULT 300,
                due_at DATETIME NULL,
                last_seen_at DATETIME NULL,
                last_synced_at DATETIME NULL,
                locked_until DATETIME NULL,
                locked_by VARCHAR(120) NULL,
                status VARCHAR(40) DEFAULT 'pending',
                last_provider VARCHAR(60) NULL,
                last_message TEXT NULL,
                last_inserted_count INT DEFAULT 0,
                last_updated_count INT DEFAULT 0,
                fail_count INT DEFAULT 0,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uniq_symbol_timeframe (symbol, timeframe),
                INDEX idx_due_priority (status, due_at, priority_score),
                INDEX idx_lock (locked_until, locked_by),
                INDEX idx_symbol_timeframe (symbol, timeframe),
                INDEX idx_last_seen (last_seen_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        ");
    } catch (Throwable $e) {
        echo "Candle sync queue table creation failed: " . $e->getMessage() . PHP_EOL;
    }
}

function sync_user_plan_sql_expression(): string
{
    if (!sync_table_exists('users')) {
        return "'free'";
    }

    foreach (['plan_code', 'subscription_plan', 'account_plan', 'membership_plan', 'plan', 'user_plan'] as $column) {
        if (sync_column_exists('users', $column)) {
            return "LOWER(COALESCE(NULLIF(TRIM(u.`{$column}`), ''), 'free'))";
        }
    }

    return "'free'";
}

function sync_is_free_plan_sql(string $planExpr): string
{
    return "({$planExpr} IN ('free','trial','demo','guest','starter_free') OR {$planExpr} LIKE 'free%')";
}

function sync_rebuild_alert_aware_queue(array $options, array $fallbackTimeframes): array
{
    sync_ensure_candle_queue_tables();

    $requestedSymbols = sync_cli_symbol_filter($options);
    $requestedAssets = sync_requested_asset_types($options);
    $maxBuildPairs = isset($options['queue-build-limit'])
        ? max(1, min(200000, (int)$options['queue-build-limit']))
        : (int)sync_setting('candle_sync_scheduler_build_pair_limit', '50000');
    $maxBuildPairs = max(1, min(200000, $maxBuildPairs));

    $stageFilterRaw = strtoupper(trim(sync_setting('candle_sync_alert_aware_stage_filter', 'LIQUIDITY_SWEPT,STRUCTURE_CONFIRMED')));
    $stageFilterAll = ($stageFilterRaw === '' || $stageFilterRaw === 'ALL' || $stageFilterRaw === '*');
    $stageFilter = $stageFilterAll ? [] : array_values(array_filter(array_map('trim', explode(',', $stageFilterRaw))));

    $symbolExpr = "UPPER(REPLACE(REPLACE(REPLACE(REPLACE(a.symbol, '/', ''), '-', ''), '_', ''), ' ', ''))";
    $planExpr = sync_user_plan_sql_expression();
    $freePlanSql = sync_is_free_plan_sql($planExpr);

    $where = [
        "a.is_active = 1",
        "a.triggered_at IS NULL",
        "(a.expired_at IS NULL OR a.expired_at = '0000-00-00 00:00:00')",
        "COALESCE(a.lifecycle_stage, 'TRACKING') NOT IN ('ALERT_SENT', 'INVALIDATED', 'EXPIRED')",
        "a.symbol IS NOT NULL",
        "a.symbol <> ''",
        "a.timeframe IS NOT NULL",
        "a.timeframe <> ''",
    ];
    $params = [];

    if (!$stageFilterAll && !empty($stageFilter)) {
        $where[] = "UPPER(COALESCE(a.lifecycle_stage, 'TRACKING')) IN (" . implode(',', array_fill(0, count($stageFilter), '?')) . ")";
        foreach ($stageFilter as $stage) {
            $params[] = $stage;
        }
    }

    if (!empty($requestedSymbols)) {
        $where[] = $symbolExpr . " IN (" . implode(',', array_fill(0, count($requestedSymbols), '?')) . ")";
        foreach ($requestedSymbols as $symbol) {
            $params[] = $symbol;
        }
    }

    $joinUsers = sync_table_exists('users') ? "LEFT JOIN users u ON u.id = a.user_id" : "";

    $stageRankSql = "MAX(CASE
        WHEN UPPER(COALESCE(a.lifecycle_stage, 'TRACKING')) IN ('FINAL_SETUP_READY','EXECUTION_READY','READY','FINAL_READY') THEN 100
        WHEN UPPER(COALESCE(a.lifecycle_stage, 'TRACKING')) IN ('STRUCTURE_CONFIRMED','STAGE_3','BOS_CONFIRMED','CHOCH_CONFIRMED') THEN 85
        WHEN UPPER(COALESCE(a.lifecycle_stage, 'TRACKING')) IN ('LIQUIDITY_SWEPT','STAGE_2') THEN 70
        WHEN UPPER(COALESCE(a.lifecycle_stage, 'TRACKING')) IN ('ZONE_APPROACHED','ZONE_REACHED','STAGE_1') THEN 35
        ELSE 15
    END)";

    try {
        $sql = "
            SELECT
                {$symbolExpr} AS symbol,
                a.timeframe,
                COUNT(*) AS alert_count,
                SUM(CASE WHEN {$freePlanSql} THEN 1 ELSE 0 END) AS free_count,
                SUM(CASE WHEN {$freePlanSql} THEN 0 ELSE 1 END) AS paid_count,
                {$stageRankSql} AS stage_rank,
                MIN(COALESCE(a.last_checked_at, a.tracking_started_at, a.created_at)) AS oldest_work
            FROM institutional_alerts a
            {$joinUsers}
            WHERE " . implode(' AND ', $where) . "
            GROUP BY {$symbolExpr}, a.timeframe
            ORDER BY stage_rank DESC, paid_count DESC, alert_count DESC, oldest_work ASC, symbol ASC, a.timeframe ASC
            LIMIT {$maxBuildPairs}
        ";
        $stmt = db()->prepare($sql);
        $stmt->execute($params);
        $pairs = $stmt->fetchAll(PDO::FETCH_ASSOC) ?: [];
    } catch (Throwable $e) {
        echo "Enterprise queue rebuild failed: " . $e->getMessage() . PHP_EOL;
        return ['pairs_seen' => 0, 'jobs_upserted' => 0, 'stale_marked' => 0, 'error' => $e->getMessage()];
    }

    $upserted = 0;
    foreach ($pairs as $pair) {
        $symbol = strtoupper(trim((string)($pair['symbol'] ?? '')));
        if ($symbol === '') continue;

        $assetType = sync_guess_asset_type($symbol);
        if ($requestedAssets && !in_array($assetType, $requestedAssets, true)) {
            continue;
        }

        $setupTf = sync_normalize_timeframe_code((string)($pair['timeframe'] ?? ''));
        if ($setupTf === '') continue;

        $needed = [$setupTf];
        $confirmTf = sync_confirmation_timeframe_for_setup($setupTf);
        if ($confirmTf !== '' && !in_array($confirmTf, $needed, true)) {
            $needed[] = $confirmTf;
        }

        if (sync_bool_option($options, 'include-requested-timeframes', false)) {
            foreach ($fallbackTimeframes as $tf) {
                $tf = sync_normalize_timeframe_code($tf);
                if ($tf !== '' && !in_array($tf, $needed, true)) {
                    $needed[] = $tf;
                }
            }
        }

        $alertCount = (int)($pair['alert_count'] ?? 0);
        $freeCount = (int)($pair['free_count'] ?? 0);
        $paidCount = (int)($pair['paid_count'] ?? 0);
        $stageRank = (int)($pair['stage_rank'] ?? 0);
        $bucket = sync_queue_priority_bucket($stageRank, $paidCount);
        $provider = $assetType === 'CRYPTO' ? 'binance' : sync_alert_aware_default_provider($assetType);

        foreach ($needed as $tf) {
            $tf = sync_normalize_timeframe_code($tf);
            if ($tf === '') continue;

            $minInterval = sync_queue_min_interval_seconds($tf, $stageRank, $paidCount, $freeCount);
            $priority = sync_queue_priority_score($stageRank, $alertCount, $paidCount, $freeCount, $tf);

            try {
                $stmt = db()->prepare("
                    INSERT INTO candle_sync_queue
                        (symbol, timeframe, asset_type, provider, priority_score, priority_bucket,
                         alert_count, free_count, paid_count, stage_rank, min_interval_seconds,
                         due_at, last_seen_at, status, last_message)
                    VALUES
                        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), NOW(), 'pending', 'Queued from active tracked setups.')
                    ON DUPLICATE KEY UPDATE
                        asset_type = VALUES(asset_type),
                        provider = VALUES(provider),
                        priority_score = VALUES(priority_score),
                        priority_bucket = VALUES(priority_bucket),
                        alert_count = VALUES(alert_count),
                        free_count = VALUES(free_count),
                        paid_count = VALUES(paid_count),
                        stage_rank = VALUES(stage_rank),
                        min_interval_seconds = VALUES(min_interval_seconds),
                        last_seen_at = NOW(),
                        status = CASE WHEN status = 'disabled' THEN status ELSE 'pending' END,
                        due_at = CASE WHEN due_at IS NULL THEN NOW() ELSE due_at END,
                        updated_at = NOW()
                ");
                $stmt->execute([$symbol, $tf, $assetType, $provider, $priority, $bucket, $alertCount, $freeCount, $paidCount, $stageRank, $minInterval]);
                $upserted++;
            } catch (Throwable $e) {
                echo "Queue upsert failed for {$symbol} {$tf}: " . $e->getMessage() . PHP_EOL;
            }
        }
    }

    $staleMarked = 0;
    try {
        $staleMinutes = max(10, min(1440, (int)sync_setting('candle_sync_scheduler_stale_after_minutes', '30')));
        $stmt = db()->prepare("
            UPDATE candle_sync_queue
            SET status = 'inactive', locked_until = NULL, locked_by = NULL, updated_at = NOW()
            WHERE last_seen_at IS NOT NULL
              AND last_seen_at < DATE_SUB(NOW(), INTERVAL ? MINUTE)
              AND status <> 'disabled'
        ");
        $stmt->execute([$staleMinutes]);
        $staleMarked = $stmt->rowCount();
    } catch (Throwable $e) {
    }

    return ['pairs_seen' => count($pairs), 'jobs_upserted' => $upserted, 'stale_marked' => $staleMarked];
}

function sync_queue_priority_bucket(int $stageRank, int $paidCount): string
{
    if ($stageRank >= 90) return $paidCount > 0 ? 'paid_final_ready' : 'free_final_ready';
    if ($stageRank >= 80) return $paidCount > 0 ? 'paid_structure_confirmed' : 'free_structure_confirmed';
    if ($stageRank >= 70) return $paidCount > 0 ? 'paid_liquidity_swept' : 'free_liquidity_swept';
    return $paidCount > 0 ? 'paid_watchlist' : 'free_watchlist';
}

function sync_queue_priority_score(int $stageRank, int $alertCount, int $paidCount, int $freeCount, string $timeframe): float
{
    $paidBoost = (float)sync_setting('candle_sync_scheduler_paid_priority_boost', '50');
    $freePenalty = (float)sync_setting('candle_sync_scheduler_free_priority_penalty', '20');
    $tfSeconds = max(60, sync_timeframe_seconds($timeframe));
    $fastTfBoost = max(0, 20 - min(20, $tfSeconds / 300));

    $score = (float)$stageRank;
    $score += min(100, $alertCount) * 0.2;
    if ($paidCount > 0) $score += $paidBoost;
    if ($paidCount <= 0 && $freeCount > 0) $score -= $freePenalty;
    $score += $fastTfBoost;

    return round(max(1, $score), 4);
}

function sync_queue_min_interval_seconds(string $timeframe, int $stageRank, int $paidCount, int $freeCount): int
{
    $base = sync_minimum_seconds_between_calls($timeframe, 'live');

    if ($stageRank < 60) {
        $base *= max(1, (int)sync_setting('candle_sync_scheduler_watchlist_interval_multiplier', '2'));
    }

    if ($paidCount <= 0 && $freeCount > 0) {
        $freeMinimum = max(300, (int)sync_setting('candle_sync_free_plan_min_seconds', '1800'));
        $freeMultiplier = max(1, (int)sync_setting('candle_sync_free_plan_interval_multiplier', '3'));
        $base = max($base * $freeMultiplier, $freeMinimum);
    }

    if ($stageRank >= 85 && $paidCount > 0) {
        $base = max(60, (int)floor($base * 0.75));
    }

    return max(60, min(86400, (int)$base));
}

function sync_queue_due_symbol_rows(string $workerId, int $limit, array $options): array
{
    sync_ensure_candle_queue_tables();
    $limit = max(1, min(10000, $limit));
    $lockSeconds = max(60, min(3600, (int)sync_setting('candle_sync_scheduler_lock_seconds', '900')));
    $requestedSymbols = sync_cli_symbol_filter($options);
    $requestedAssets = sync_requested_asset_types($options);

    $where = [
        "status IN ('pending','retry','ok','cache_skipped','no_data','error','provider_limit','provider_cooldown')",
        "(due_at IS NULL OR due_at <= NOW())",
        "(locked_until IS NULL OR locked_until <= NOW())",
    ];
    $params = [];

    if (!empty($requestedSymbols)) {
        $where[] = "symbol IN (" . implode(',', array_fill(0, count($requestedSymbols), '?')) . ")";
        foreach ($requestedSymbols as $symbol) $params[] = $symbol;
    }

    if (!empty($requestedAssets)) {
        $where[] = "asset_type IN (" . implode(',', array_fill(0, count($requestedAssets), '?')) . ")";
        foreach ($requestedAssets as $asset) $params[] = $asset;
    }

    try {
        $sql = "
            SELECT id
            FROM candle_sync_queue
            WHERE " . implode(' AND ', $where) . "
            ORDER BY priority_score DESC, COALESCE(due_at, '1970-01-01 00:00:00') ASC, id ASC
            LIMIT {$limit}
        ";
        $stmt = db()->prepare($sql);
        $stmt->execute($params);
        $ids = array_map('intval', $stmt->fetchAll(PDO::FETCH_COLUMN) ?: []);
        $ids = array_values(array_filter($ids));
        if (!$ids) return [];

        $placeholders = implode(',', array_fill(0, count($ids), '?'));
        $updateParams = array_merge([$workerId, $lockSeconds], $ids);
        $stmt = db()->prepare("
            UPDATE candle_sync_queue
            SET locked_by = ?, locked_until = DATE_ADD(NOW(), INTERVAL ? SECOND), updated_at = NOW()
            WHERE id IN ({$placeholders})
              AND (locked_until IS NULL OR locked_until <= NOW() OR locked_by = ?)
        ");
        $updateParams[] = $workerId;
        $stmt->execute($updateParams);

        $stmt = db()->prepare("
            SELECT *
            FROM candle_sync_queue
            WHERE id IN ({$placeholders})
              AND locked_by = ?
            ORDER BY priority_score DESC, COALESCE(due_at, '1970-01-01 00:00:00') ASC, id ASC
        ");
        $stmt->execute(array_merge($ids, [$workerId]));
        $jobs = $stmt->fetchAll(PDO::FETCH_ASSOC) ?: [];
    } catch (Throwable $e) {
        echo "Queue select failed: " . $e->getMessage() . PHP_EOL;
        return [];
    }

    $rows = [];
    foreach ($jobs as $job) {
        $symbol = strtoupper(trim((string)($job['symbol'] ?? '')));
        $tf = sync_normalize_timeframe_code((string)($job['timeframe'] ?? ''));
        if ($symbol === '' || $tf === '') continue;

        if (!isset($rows[$symbol])) {
            $rows[$symbol] = [
                'symbol' => $symbol,
                'asset_type' => sync_row_asset_type($job, $symbol),
                'provider' => strtolower(trim((string)($job['provider'] ?? ''))),
                '__sync_timeframes' => [],
                '__sync_alert_aware' => true,
                '__sync_queue' => true,
                '__queue_job_ids' => [],
                '__sync_alert_count' => 0,
            ];
        }

        if (!in_array($tf, $rows[$symbol]['__sync_timeframes'], true)) {
            $rows[$symbol]['__sync_timeframes'][] = $tf;
        }
        $rows[$symbol]['__queue_job_ids'][$tf] = (int)$job['id'];
        $rows[$symbol]['__sync_alert_count'] += (int)($job['alert_count'] ?? 0);
    }

    foreach ($rows as &$row) {
        usort($row['__sync_timeframes'], 'sync_timeframe_sort_compare');
    }
    unset($row);

    return array_values($rows);
}

function sync_queue_finish(int $jobId, string $status, string $provider, int $inserted, int $updated, string $message, int $rescheduleSeconds): void
{
    if ($jobId <= 0) return;

    $status = strtolower(preg_replace('/[^a-z0-9_\-]/i', '_', $status));
    $provider = strtolower(trim($provider));
    $rescheduleSeconds = max(30, min(86400, $rescheduleSeconds));
    $message = sync_short_log_message($message);

    try {
        $failInc = in_array($status, ['ok','cache_skipped'], true) ? '0' : '1';
        $lastSyncedSql = $status === 'ok' ? 'NOW()' : 'last_synced_at';
        $stmt = db()->prepare("
            UPDATE candle_sync_queue
            SET status = ?,
                last_provider = ?,
                last_message = ?,
                last_inserted_count = ?,
                last_updated_count = ?,
                last_synced_at = {$lastSyncedSql},
                due_at = DATE_ADD(NOW(), INTERVAL ? SECOND),
                locked_until = NULL,
                locked_by = NULL,
                fail_count = CASE WHEN {$failInc} = 1 THEN COALESCE(fail_count,0) + 1 ELSE 0 END,
                updated_at = NOW()
            WHERE id = ?
        ");
        $stmt->execute([$status, $provider, $message, $inserted, $updated, $rescheduleSeconds, $jobId]);
    } catch (Throwable $e) {
        // Queue bookkeeping must not stop candle syncing.
    }
}

function sync_api_delay_microseconds(string $provider): int
{
    $provider = strtolower(trim($provider));

    if ($provider === 'binance') {
        return 200000;
    }

    if ($provider === 'fmp' || $provider === 'financialmodelingprep') {
        return 400000;
    }

    return 500000;
}
