qsmarketdata.lua - main - CountZero/quik-cscalp-lua

qsmarketdata.lua
2 173 строки79 KB
CountZero
Update
05 авг 2026, 20:59
05 авг 2026, 20:59500bdaf
-- Ownership and recovery of Level II, tick and candle subscriptions.

local M = {}
M.__index = M

local TICK_FIELDS = {
    "trade_num", "flags", "price", "qty", "value", "sec_code",
    "class_code", "datetime", "period", "open_interest", "exchange_code",
    "exec_market",
}

local TICK_STARTUP_PROBE_MSEC = 12000
local TICK_STARTUP_RECREATE_LIMIT = 1
local TICK_DIAGNOSTIC_FAST_MSEC = 15000
local TICK_DIAGNOSTIC_SLOW_MSEC = 60000
local TICK_DIAGNOSTIC_FAST_LIMIT = 4
-- INTERVAL_TICK may replay the local day after subscription. Only near-live
-- trades may enter the bounded CScalp callback queue.
local TICK_REALTIME_GRACE_SEC = 3
local MARKET_EVENT_QUEUE_LIMIT = 32768
local MARKET_EVENT_FLUSH_LIMIT = 512

local function trim(value)
    if value == nil then
        return ""
    end
    return (tostring(value):gsub("^%s+", ""):gsub("%s+$", ""))
end

local function normalize_class(value)
    return string.upper(trim(value))
end

local function key2(class_code, sec_code)
    return normalize_class(class_code) .. "|" .. trim(sec_code)
end

local function key3(class_code, sec_code, interval)
    return key2(class_code, sec_code) .. "|" .. tostring(interval)
end

local function owner_key(value)
    local owner = trim(value)
    if owner == "" then
        return "__legacy_instrument_owner__"
    end
    return owner
end

local function owner_count(record)
    local count = 0
    for _, _ in pairs(record.owners or {}) do
        count = count + 1
    end
    return count
end

local function sorted_keys(value)
    local result = {}
    for key, _ in pairs(value or {}) do
        result[#result + 1] = key
    end
    table.sort(result)
    return result
end

local function shallow_copy(value)
    if type(value) ~= "table" then
        return value
    end
    local result = {}
    for key, item in pairs(value) do
        result[key] = item
    end
    return result
end


local function market_event_depth(self)
    local depth = self.market_event_tail - self.market_event_head + 1
    if depth < 0 then
        return 0
    end
    return depth
end

local function reset_market_events(self)
    self.market_events = {}
    self.market_event_head = 1
    self.market_event_tail = 0
end

local function discard_market_events(self, reason)
    local depth = market_event_depth(self)
    if depth == 0 then
        reset_market_events(self)
        return 0
    end
    self.metrics.market_events_discarded =
        self.metrics.market_events_discarded + depth
    reset_market_events(self)
    self.log("Queued market events discarded; count=" .. tostring(depth)
        .. "; reason=" .. tostring(reason), "warn")
    return depth
end

local function should_log_counter(value)
    if value <= 1 then
        return true
    end
    local power = 1
    while power < value do
        power = power * 2
    end
    return power == value
end

local function enqueue_market_event(self, command, data)
    local depth = market_event_depth(self)
    if depth >= MARKET_EVENT_QUEUE_LIMIT then
        self.metrics.market_event_overflows =
            self.metrics.market_event_overflows + 1
        if should_log_counter(self.metrics.market_event_overflows) then
            self.log("Market event queue overflow; command="
                .. tostring(command) .. "; depth=" .. tostring(depth)
                .. "; dropped="
                .. tostring(self.metrics.market_event_overflows), "error")
        end
        return nil, "market event queue overflow"
    end

    local tail = self.market_event_tail + 1
    self.market_event_tail = tail
    self.market_events[tail] = {
        command = command,
        data = data,
    }
    depth = depth + 1
    self.metrics.market_events_enqueued =
        self.metrics.market_events_enqueued + 1
    if depth > self.metrics.market_event_peak then
        self.metrics.market_event_peak = depth
    end
    return true
end

local function flush_market_events(self)
    local emitted = 0
    while self.market_event_head <= self.market_event_tail
        and emitted < MARKET_EVENT_FLUSH_LIMIT do
        local head = self.market_event_head
        local event = self.market_events[head]
        if not event then
            self.market_event_head = head + 1
        else
            local ok, emit_error = self.emit(event.command, event.data)
            if not ok then
                self.metrics.market_event_emit_failures =
                    self.metrics.market_event_emit_failures + 1
                self.log("Market event delivery paused; command="
                    .. tostring(event.command) .. "; error="
                    .. tostring(emit_error), "warn")
                return false
            end
            self.market_events[head] = nil
            self.market_event_head = head + 1
            emitted = emitted + 1
            self.metrics.market_events_emitted =
                self.metrics.market_events_emitted + 1
        end
    end
    if self.market_event_head > self.market_event_tail then
        reset_market_events(self)
    end
    return true
end

local function default_api()
    return {
        subscribe_l2 = function(class_code, sec_code)
            return _G.Subscribe_Level_II_Quotes(class_code, sec_code)
        end,
        unsubscribe_l2 = function(class_code, sec_code)
            return _G.Unsubscribe_Level_II_Quotes(class_code, sec_code)
        end,
        is_subscribed_l2 = function(class_code, sec_code)
            return _G.IsSubscribed_Level_II_Quotes(class_code, sec_code)
        end,
        get_quote = function(class_code, sec_code)
            return _G.getQuoteLevel2(class_code, sec_code)
        end,
        get_info = function(name)
            return _G.getInfoParam(name)
        end,
        create_data_source = function(class_code, sec_code, interval)
            return _G.CreateDataSource(class_code, sec_code, interval)
        end,
        get_number_of = function(table_name)
            if type(_G.getNumberOf) ~= "function" then
                return nil, "getNumberOf is unavailable"
            end
            return _G.getNumberOf(table_name)
        end,
        get_item = function(table_name, index)
            if type(_G.getItem) ~= "function" then
                return nil, "getItem is unavailable"
            end
            return _G.getItem(table_name, index)
        end,
        get_server_datetime = function()
            if type(_G.getInfoParam) ~= "function" then
                return nil, nil, "getInfoParam is unavailable"
            end
            return _G.getInfoParam("TRADEDATE"),
                _G.getInfoParam("SERVERTIME")
        end,
        interval_tick = function()
            return rawget(_G, "INTERVAL_TICK") or 0
        end,
    }
end

local function merge_api(custom)
    local result = default_api()
    for key, value in pairs(custom or {}) do
        result[key] = value
    end
    return result
end

function M.new(config, options)
    options = options or {}
    return setmetatable({
        config = config,
        now = options.now or function() return 0 end,
        delay = options.delay or function() end,
        pump = options.pump or function() end,
        log = options.log or function() end,
        emit = options.emit or function() return true end,
        is_ready = options.is_ready or function() return false end,
        api = merge_api(options.api),
        books = {},
        candles = {},
        pending_params = {},
        param_sequence = 0,
        market_events = {},
        market_event_head = 1,
        market_event_tail = 0,
        server_time = { value = nil, expires_msec = 0 },
        recovering = false,
        recovery_books = {},
        recovery_book_i = 1,
        recovery_candles = {},
        recovery_candle_i = 1,
        last_market_event_msec = 0,
        last_l2_audit_msec = 0,
        last_native_all_trades = {
            count = nil,
            error = "not probed",
        },
        metrics = {
            all_trade_forwarded = 0,
            all_trade_dropped = 0,
            all_trade_callbacks_received = 0,
            all_trade_unmatched = 0,
            all_trade_replay_dropped = 0,
            all_trade_datetime_invalid = 0,
            tick_replay_catchups = 0,
            quote_forwarded = 0,
            quote_dropped = 0,
            param_forwarded = 0,
            param_dropped = 0,
            param_coalesced = 0,
            tick_create_failures = 0,
            candle_create_failures = 0,
            candle_dropped = 0,
            retry_attempts = 0,
            stale_generation_callbacks = 0,
            l2_audit_failures = 0,
            tick_confirmations = 0,
            tick_startup_stalls = 0,
            tick_recreates = 0,
            tick_recreate_failures = 0,
            tick_probe_failures = 0,
            tick_update_callbacks = 0,
            tick_callback_downgrades = 0,
            tick_callback_downgrade_failures = 0,
            tick_close_attempts = 0,
            tick_close_successes = 0,
            tick_close_failures = 0,
            tick_diagnostic_snapshots = 0,
            native_all_trades_probe_failures = 0,
            market_events_enqueued = 0,
            market_events_emitted = 0,
            market_event_emit_failures = 0,
            market_event_overflows = 0,
            market_events_discarded = 0,
            market_event_peak = 0,
        },
    }, M)
end

function M:is_supported(class_code)
    return self.config.class_set[normalize_class(class_code)] == true
end

local function safe_close(self, data_source, reason)
    if not data_source then
        return true
    end
    local close_method = data_source.Close
    if type(close_method) ~= "function" then
        self.log("DataSource has no Close method: " .. tostring(reason), "warn")
        return nil
    end
    local ok, close_result = pcall(close_method, data_source)
    if not ok or close_result == false then
        self.log("DataSource close failed: " .. tostring(reason)
            .. "; error=" .. tostring(close_result), "warn")
        return nil
    end
    return true
end

local function close_tick_source(self, record, reason)
    local data_source = record and record.tick_handle or nil
    if not data_source then
        return true
    end
    self.metrics.tick_close_attempts =
        self.metrics.tick_close_attempts + 1
    self.log("Tick DataSource close requested: " .. record.key
        .. "; generation=" .. tostring(record.tick_generation)
        .. "; state=" .. tostring(record.tick_state)
        .. "; callback_mode=" .. tostring(record.tick_callback_mode)
        .. "; update_callbacks="
        .. tostring(record.tick_callback_count or 0)
        .. "; generation_all_trades="
        .. tostring(record.tick_all_trade_count or 0)
        .. "; generation_forwarded="
        .. tostring(record.tick_forwarded_count or 0)
        .. "; replay_suppressed="
        .. tostring(record.tick_replay_dropped or 0)
        .. "; reason=" .. tostring(reason), "info")
    local closed = safe_close(self, data_source,
        tostring(reason) .. " tick " .. record.key)
    if closed then
        self.metrics.tick_close_successes =
            self.metrics.tick_close_successes + 1
        self.log("Tick DataSource closed: " .. record.key
            .. "; generation=" .. tostring(record.tick_generation)
            .. "; reason=" .. tostring(reason), "info")
        return true
    end
    self.metrics.tick_close_failures =
        self.metrics.tick_close_failures + 1
    self.log("Tick DataSource remains owned after Close failure: "
        .. record.key .. "; generation="
        .. tostring(record.tick_generation)
        .. "; reason=" .. tostring(reason), "error")
    return nil
end

local function schedule_retry(self, record, kind, error_text)
    record.retry_attempt = (record.retry_attempt or 0) + 1
    local base = kind == "tick" and 1000 or 3000
    local exponent = math.min(record.retry_attempt - 1, 5)
    local delay_msec = math.min(30000, base * (2 ^ exponent))
    local jitter = (#record.key * 37 + record.retry_attempt * 101) % 251
    record.retry_due_msec = self.now() + delay_msec + jitter
    record.last_error = tostring(error_text)
    self.metrics.retry_attempts = self.metrics.retry_attempts + 1
end

local function native_all_trades_snapshot(self)
    local snapshot = {
        count = nil,
        error = nil,
        last_key = nil,
        last_trade_num = nil,
    }
    local count_ok, count, count_error = pcall(
        self.api.get_number_of, "all_trades")
    if not count_ok then
        snapshot.error = tostring(count)
    elseif tonumber(count) == nil or tonumber(count) < 0 then
        snapshot.error = tostring(count_error
            or "getNumberOf(all_trades) returned " .. tostring(count))
    else
        snapshot.count = math.floor(tonumber(count))
    end

    if snapshot.count and snapshot.count > 0 then
        local item_ok, item, item_error = pcall(
            self.api.get_item, "all_trades", snapshot.count - 1)
        if item_ok and type(item) == "table" then
            snapshot.last_key = key2(item.class_code, item.sec_code)
            snapshot.last_trade_num = item.trade_num
        elseif not item_ok then
            snapshot.error = "getItem(all_trades) failed: "
                .. tostring(item)
        elseif item_error ~= nil then
            snapshot.error = "getItem(all_trades) failed: "
                .. tostring(item_error)
        end
    end

    if snapshot.error then
        self.metrics.native_all_trades_probe_failures =
            self.metrics.native_all_trades_probe_failures + 1
    end
    self.last_native_all_trades = snapshot
    return snapshot
end

local function native_snapshot_text(snapshot)
    if not snapshot then
        return "native_all_trades=not_probed"
    end
    local text = "native_all_trades_count="
        .. tostring(snapshot.count)
    if snapshot.last_key then
        text = text .. "; native_last_key=" .. snapshot.last_key
            .. "; native_last_trade_num="
            .. tostring(snapshot.last_trade_num)
    end
    if snapshot.error then
        text = text .. "; native_probe_error="
            .. tostring(snapshot.error)
    end
    return text
end

local function valid_datetime_parts(year, month, day, hour, minute, second)
    return year and year >= 2000 and year <= 2200
        and month and month >= 1 and month <= 12
        and day and day >= 1 and day <= 31
        and hour and hour >= 0 and hour <= 23
        and minute and minute >= 0 and minute <= 59
        and second and second >= 0 and second <= 60
end

local function datetime_epoch(year, month, day, hour, minute, second)
    year = tonumber(year)
    month = tonumber(month)
    day = tonumber(day)
    hour = tonumber(hour)
    minute = tonumber(minute)
    second = tonumber(second)
    if not valid_datetime_parts(
        year, month, day, hour, minute, second) then
        return nil, "invalid datetime fields"
    end
    local ok, value = pcall(os.time, {
        year = year,
        month = month,
        day = day,
        hour = hour,
        min = minute,
        sec = second,
    })
    if not ok or tonumber(value) == nil then
        return nil, tostring(value or "os.time returned nil")
    end
    return tonumber(value)
end

local function parse_server_datetime(date_text, time_text)
    local date = trim(date_text)
    local time = trim(time_text)
    local day, month, year = date:match(
        "^(%d%d?)%.(%d%d?)%.(%d%d%d%d)$")
    if not day then
        year, month, day = date:match(
            "^(%d%d%d%d)[%./%-](%d%d?)[%./%-](%d%d?)$")
    end
    if not day then
        day, month, year = date:match(
            "^(%d%d?)[/%-](%d%d?)[/%-](%d%d%d%d)$")
    end
    if not day and date:match("^%d%d%d%d%d%d%d%d$") then
        year = date:sub(1, 4)
        month = date:sub(5, 6)
        day = date:sub(7, 8)
    end
    local hour, minute, second = time:match(
        "^(%d%d?):(%d%d?):(%d%d?)")
    if not day or not hour then
        return nil, "unsupported TRADEDATE/SERVERTIME format: date="
            .. tostring(date) .. "; time=" .. tostring(time)
    end
    return datetime_epoch(year, month, day, hour, minute, second)
end

local function all_trade_epoch(record, datetime)
    if type(datetime) ~= "table" then
        return nil, "datetime is not a table"
    end
    local cache_key = table.concat({
        tostring(datetime.year), tostring(datetime.month),
        tostring(datetime.day), tostring(datetime.hour),
        tostring(datetime.min), tostring(datetime.sec),
    }, "|")
    if cache_key == record.tick_last_datetime_key then
        return record.tick_last_datetime_epoch
    end
    local value, value_error = datetime_epoch(
        datetime.year, datetime.month, datetime.day,
        datetime.hour, datetime.min, datetime.sec)
    if value then
        record.tick_last_datetime_key = cache_key
        record.tick_last_datetime_epoch = value
    end
    return value, value_error
end

local function reset_tick_replay_state(record)
    record.tick_replay_filter_enabled = false
    record.tick_replay_clock_epoch = nil
    record.tick_replay_clock_msec = nil
    record.tick_replay_clock_error = nil
    record.tick_replay_dropped = 0
    record.tick_replay_caught_up = false
    record.tick_replay_max_age_sec = 0
    record.tick_last_datetime_key = nil
    record.tick_last_datetime_epoch = nil
end

local function initialize_replay_filter(self, record)
    reset_tick_replay_state(record)

    local ok, date_text, time_text, api_error = pcall(
        self.api.get_server_datetime)
    if not ok then
        record.tick_replay_clock_error = tostring(date_text)
        return nil
    end
    local epoch, epoch_error = parse_server_datetime(date_text, time_text)
    if not epoch then
        record.tick_replay_clock_error = tostring(
            api_error or epoch_error)
        return nil
    end
    record.tick_replay_filter_enabled = true
    record.tick_replay_clock_epoch = epoch
    record.tick_replay_clock_msec = self.now()
    return true
end

local function estimated_server_epoch(self, record)
    if not record.tick_replay_filter_enabled
        or not record.tick_replay_clock_epoch
        or not record.tick_replay_clock_msec then
        return nil
    end
    local elapsed_msec = math.max(
        0, self.now() - record.tick_replay_clock_msec)
    return record.tick_replay_clock_epoch
        + math.floor(elapsed_msec / 1000)
end

local function suppress_historical_all_trade(self, record, all_trade)
    if not record.tick_replay_filter_enabled then
        return false
    end
    local trade_epoch, trade_error = all_trade_epoch(
        record, all_trade.datetime)
    if not trade_epoch then
        self.metrics.all_trade_datetime_invalid =
            self.metrics.all_trade_datetime_invalid + 1
        self.metrics.all_trade_dropped =
            self.metrics.all_trade_dropped + 1
        if should_log_counter(
            self.metrics.all_trade_datetime_invalid) then
            self.log("OnAllTrade dropped because datetime cannot be "
                .. "validated while replay protection is active: "
                .. record.key .. "; trade_num="
                .. tostring(all_trade.trade_num)
                .. "; error=" .. tostring(trade_error)
                .. "; invalid_datetime="
                .. tostring(self.metrics.all_trade_datetime_invalid),
                "info")
        end
        return true
    end

    local current_epoch = estimated_server_epoch(self, record)
    if not current_epoch then
        return false
    end
    local age_sec = current_epoch - trade_epoch
    if age_sec > TICK_REALTIME_GRACE_SEC then
        record.tick_replay_dropped =
            (record.tick_replay_dropped or 0) + 1
        record.tick_replay_max_age_sec = math.max(
            record.tick_replay_max_age_sec or 0, age_sec)
        self.metrics.all_trade_replay_dropped =
            self.metrics.all_trade_replay_dropped + 1
        self.metrics.all_trade_dropped =
            self.metrics.all_trade_dropped + 1
        if should_log_counter(record.tick_replay_dropped) then
            self.log("Historical OnAllTrade suppressed: " .. record.key
                .. "; generation="
                .. tostring(record.tick_generation)
                .. "; dropped="
                .. tostring(record.tick_replay_dropped)
                .. "; trade_num=" .. tostring(all_trade.trade_num)
                .. "; age_sec=" .. tostring(age_sec)
                .. "; grace_sec="
                .. tostring(TICK_REALTIME_GRACE_SEC)
                .. "; queue_depth="
                .. tostring(market_event_depth(self)), "info")
        end
        return true
    end

    if not record.tick_replay_caught_up then
        record.tick_replay_caught_up = true
        self.metrics.tick_replay_catchups =
            self.metrics.tick_replay_catchups + 1
        self.log("Tick replay caught up to realtime: " .. record.key
            .. "; generation=" .. tostring(record.tick_generation)
            .. "; suppressed="
            .. tostring(record.tick_replay_dropped or 0)
            .. "; max_age_sec="
            .. tostring(record.tick_replay_max_age_sec or 0)
            .. "; first_live_trade_num="
            .. tostring(all_trade.trade_num)
            .. "; live_age_sec=" .. tostring(age_sec), "info")
    end
    return false
end

local function set_tick_callback(self, record, data_source, generation)
    local errors = {}
    if type(data_source.SetUpdateCallback) == "function" then
        local callback = function(index)
            if record.tick_generation ~= generation
                or record.tick_handle ~= data_source then
                self.metrics.stale_generation_callbacks =
                    self.metrics.stale_generation_callbacks + 1
                return
            end
            record.tick_callback_count =
                (record.tick_callback_count or 0) + 1
            if record.tick_callback_count == 1 then
                record.tick_first_callback_index = index
                record.tick_first_callback_msec = self.now()
            end
            record.tick_last_callback_index = index
            record.tick_last_callback_msec = self.now()
            record.tick_callback_log_pending = true
            self.metrics.tick_update_callbacks =
                self.metrics.tick_update_callbacks + 1
        end
        local ok, result = pcall(
            data_source.SetUpdateCallback, data_source, callback)
        if ok and result ~= false then
            return true, "update"
        end
        errors[#errors + 1] = ok
            and "SetUpdateCallback returned false"
            or "SetUpdateCallback failed: " .. tostring(result)
    end
    if type(data_source.SetEmptyCallback) == "function" then
        local ok, result = pcall(
            data_source.SetEmptyCallback, data_source)
        if ok and result ~= false then
            return true, "empty"
        end
        errors[#errors + 1] = ok
            and "SetEmptyCallback returned false"
            or "SetEmptyCallback failed: " .. tostring(result)
    end
    if #errors == 0 then
        return nil, "tick DataSource has no callback method"
    end
    return nil, table.concat(errors, "; ")
end

local function data_source_size(data_source)
    if not data_source or type(data_source.Size) ~= "function" then
        return nil, "DataSource has no Size method"
    end
    local ok, size = pcall(data_source.Size, data_source)
    if not ok then
        return nil, tostring(size)
    end
    return tonumber(size) or 0
end

local function confirm_tick_source(self, record, size, reason)
    if record.tick_state == "confirmed" then
        return true
    end
    record.tick_state = "confirmed"
    record.tick_probe_due_msec = nil
    record.tick_diagnostic_due_msec = nil
    record.tick_last_size = tonumber(size) or record.tick_last_size or 0
    record.tick_initial_size = nil
    record.tick_recreate_attempts = 0
    if record.tick_callback_mode == "update" then
        record.tick_callback_downgrade_pending = true
    end
    self.metrics.tick_confirmations =
        self.metrics.tick_confirmations + 1
    self.log("Tick subscription confirmed: " .. record.key
        .. "; generation=" .. tostring(record.tick_generation)
        .. "; size=" .. tostring(record.tick_last_size)
        .. "; reason=" .. tostring(reason), "info")
    return true
end

local function create_tick_source(self, record)
    if record.tick_handle then
        return true
    end
    local create_started_msec = self.now()
    local interval = self.api.interval_tick()
    local native_before = native_all_trades_snapshot(self)
    self.log("Tick DataSource creation requested: " .. record.key
        .. "; owners=" .. tostring(owner_count(record))
        .. "; interval=" .. tostring(interval)
        .. "; " .. native_snapshot_text(native_before), "info")
    local ok, data_source, create_error = pcall(
        self.api.create_data_source,
        record.class_code,
        record.sec_code,
        interval)
    if not ok then
        return nil, tostring(data_source)
    end
    if not data_source then
        return nil, tostring(create_error or "CreateDataSource returned nil")
    end
    if create_error ~= nil and tostring(create_error) ~= "" then
        safe_close(self, data_source, "tick creation returned an error")
        return nil, tostring(create_error)
    end
    local generation = (record.tick_generation or 0) + 1
    record.tick_generation = generation
    record.tick_handle = data_source
    record.tick_callback_count = 0
    record.tick_all_trade_count = 0
    record.tick_forwarded_count = 0
    record.tick_last_callback_index = nil
    record.tick_last_callback_msec = nil
    record.tick_first_callback_index = nil
    record.tick_first_callback_msec = nil
    record.tick_callback_logged_count = nil
    record.tick_callback_log_pending = false
    record.tick_callback_downgrade_pending = false
    record.tick_diagnostic_count = 0
    record.tick_diagnostic_due_msec = nil
    initialize_replay_filter(self, record)
    local callback_call_ok, callback_ok, callback_error =
        pcall(set_tick_callback, self, record, data_source, generation)
    if not callback_call_ok or callback_ok ~= true then
        record.tick_handle = nil
        safe_close(self, data_source, "tick callback setup failed")
        return nil, tostring(callback_call_ok
            and callback_error or callback_ok)
    end
    record.tick_callback_mode = callback_error
    record.tick_created_msec = self.now()
    record.tick_probe_due_msec =
        record.tick_created_msec + TICK_STARTUP_PROBE_MSEC
    record.tick_state = "warming"

    local size, size_error = data_source_size(data_source)
    record.tick_initial_size = size
    record.tick_last_size = size
    if size == nil then
        record.tick_state = "unconfirmed"
        record.tick_probe_due_msec = nil
        record.tick_diagnostic_due_msec =
            record.tick_created_msec + TICK_DIAGNOSTIC_FAST_MSEC
        self.metrics.tick_probe_failures =
            self.metrics.tick_probe_failures + 1
        self.log("Tick subscription cannot be audited: " .. record.key
            .. "; generation=" .. tostring(record.tick_generation)
            .. "; callback_mode="
            .. tostring(record.tick_callback_mode)
            .. "; error=" .. tostring(size_error)
            .. "; " .. native_snapshot_text(
                native_all_trades_snapshot(self)), "warn")
    else
        local native_after = native_all_trades_snapshot(self)
        self.log("Tick subscription warming up: " .. record.key
            .. "; generation=" .. tostring(record.tick_generation)
            .. "; callback_mode="
            .. tostring(record.tick_callback_mode)
            .. "; initial_size=" .. tostring(size)
            .. "; create_elapsed_msec="
            .. tostring(record.tick_created_msec - create_started_msec)
            .. "; replay_filter="
            .. tostring(record.tick_replay_filter_enabled)
            .. "; replay_clock_error="
            .. tostring(record.tick_replay_clock_error)
            .. "; probe_due_msec="
            .. tostring(record.tick_probe_due_msec)
            .. "; " .. native_snapshot_text(native_after), "info")
    end
    return true
end

local function activate_book(self, record)
    if not record.desired then
        return nil, "book is no longer desired"
    end
    if not self.is_ready() then
        schedule_retry(self, record, "tick", "bridge is not ready")
        return nil, "bridge is not ready"
    end

    if not record.l2_active then
        local ok, subscribed = pcall(
            self.api.subscribe_l2, record.class_code, record.sec_code)
        if not ok or not subscribed then
            local error_text = ok and "Subscribe_Level_II_Quotes returned false"
                or tostring(subscribed)
            schedule_retry(self, record, "tick", error_text)
            self.log("Level II subscription failed for " .. record.key
                .. "; retry_due_msec=" .. tostring(record.retry_due_msec)
                .. "; error=" .. tostring(error_text), "warn")
            return nil, error_text
        end
        record.l2_active = true
    end

    local tick_ok, tick_error = create_tick_source(self, record)
    if not tick_ok then
        self.metrics.tick_create_failures =
            self.metrics.tick_create_failures + 1
        schedule_retry(self, record, "tick", tick_error)
        self.log("Tick subscription failed for " .. record.key
            .. "; retry_due_msec=" .. tostring(record.retry_due_msec)
            .. "; error=" .. tostring(tick_error), "warn")
        return nil, tick_error
    end

    record.retry_attempt = 0
    record.retry_due_msec = nil
    record.last_error = nil
    record.retired_msec = nil
    record.next_l2_audit_msec = self.now() + 2000
    self.log("Market subscription initialized: " .. record.key
        .. "; tick_state=" .. tostring(record.tick_state), "info")
    return true
end

function M:subscribe_book(class_code, sec_code, consumer_id)
    local class = normalize_class(class_code)
    local security = trim(sec_code)
    if not self:is_supported(class) then
        return nil, "unsupported class: " .. class
    end
    if security == "" then
        return nil, "security code is required"
    end

    local key = key2(class, security)
    local record = self.books[key]
    local created = record == nil
    if not record then
        record = {
            key = key,
            class_code = class,
            sec_code = security,
            desired = true,
            l2_active = false,
            tick_generation = 0,
            tick_recreate_attempts = 0,
            all_trade_count = 0,
            retry_attempt = 0,
            owners = {},
        }
        self.books[key] = record
    end
    record.owners = record.owners or {}
    local owner = owner_key(consumer_id)
    local duplicate_owner = record.owners[owner] == true
    record.owners[owner] = true
    record.desired = true
    record.retired_msec = nil
    self.log("Market subscribe requested: " .. key
        .. "; consumer=" .. tostring(owner)
        .. "; owners=" .. tostring(owner_count(record))
        .. "; new_record=" .. tostring(created)
        .. "; duplicate_owner=" .. tostring(duplicate_owner)
        .. "; l2_active=" .. tostring(record.l2_active)
        .. "; tick_handle=" .. tostring(record.tick_handle ~= nil)
        .. "; tick_state=" .. tostring(record.tick_state), "info")
    if record.l2_active and record.tick_handle then
        return true
    end
    return activate_book(self, record)
end

function M:unsubscribe_book(class_code, sec_code, consumer_id)
    local key = key2(class_code, sec_code)
    local record = self.books[key]
    if not record then
        self.log("Market unsubscribe ignored; record is absent: " .. key
            .. "; consumer=" .. tostring(owner_key(consumer_id)), "info")
        return true
    end
    record.owners = record.owners or {}
    local owner = owner_key(consumer_id)
    local owner_existed = record.owners[owner] == true
    record.owners[owner] = nil
    if next(record.owners) ~= nil then
        self.log("Market unsubscribe retained shared resource: " .. key
            .. "; consumer=" .. tostring(owner)
            .. "; owner_existed=" .. tostring(owner_existed)
            .. "; remaining_owners="
            .. tostring(owner_count(record)), "info")
        return true
    end
    record.desired = false
    record.retry_due_msec = nil
    record.retired_msec = self.now()
    self.log("Market unsubscribe scheduled resource retirement: " .. key
        .. "; consumer=" .. tostring(owner)
        .. "; owner_existed=" .. tostring(owner_existed)
        .. "; retired_msec=" .. tostring(record.retired_msec)
        .. "; tick_state=" .. tostring(record.tick_state)
        .. "; tick_callbacks="
        .. tostring(record.tick_callback_count or 0)
        .. "; all_trades=" .. tostring(record.all_trade_count or 0), "info")
    return true
end

function M:is_subscribed_book(class_code, sec_code, consumer_id)
    local record = self.books[key2(class_code, sec_code)]
    local owner = consumer_id ~= nil
        and owner_key(consumer_id) or nil
    if self.is_ready() and record and record.desired
        and record.l2_active then
        local ok, actual = pcall(
            self.api.is_subscribed_l2,
            record.class_code,
            record.sec_code)
        if ok and actual == false then
            record.l2_active = false
            schedule_retry(self, record, "tick",
                "Level II subscription is not active")
            self.metrics.l2_audit_failures =
                self.metrics.l2_audit_failures + 1
        end
    end
    return self.is_ready() and record ~= nil and record.desired
        and (owner == nil or record.owners[owner] == true)
        and record.l2_active and record.tick_handle ~= nil
end

local function another_active_book(self, except_key)
    for key, record in pairs(self.books) do
        if key ~= except_key and record.desired
            and record.l2_active and record.tick_handle then
            return true
        end
    end
    return false
end

local function another_desired_book(self, except_key)
    for key, record in pairs(self.books) do
        if key ~= except_key and record.desired then
            return true
        end
    end
    return false
end

local function close_book_record(self, record, reason)
    if not close_tick_source(self, record, reason) then
        return nil
    end
    record.tick_handle = nil
    record.tick_state = nil
    record.tick_created_msec = nil
    record.tick_probe_due_msec = nil
    record.tick_diagnostic_due_msec = nil
    record.tick_diagnostic_count = 0
    record.tick_initial_size = nil
    record.tick_last_size = nil
    record.tick_callback_mode = nil
    record.tick_callback_count = 0
    record.tick_all_trade_count = 0
    record.tick_forwarded_count = 0
    record.tick_last_callback_index = nil
    record.tick_last_callback_msec = nil
    record.tick_first_callback_index = nil
    record.tick_first_callback_msec = nil
    record.tick_callback_logged_count = nil
    record.tick_callback_log_pending = false
    record.tick_callback_downgrade_pending = false
    record.tick_recreate_attempts = 0
    reset_tick_replay_state(record)
    if record.l2_active then
        local ok, unsubscribe_result = pcall(
            self.api.unsubscribe_l2, record.class_code, record.sec_code)
        if not ok or unsubscribe_result == false then
            self.log("Level II unsubscribe failed for " .. record.key
                .. "; error=" .. tostring(unsubscribe_result), "warn")
            return nil
        end
    end
    record.l2_active = false
    self.log("Market subscription resources released: " .. record.key
        .. "; reason=" .. tostring(reason), "info")
    return true
end

function M:on_all_trade(all_trade)
    self.metrics.all_trade_callbacks_received =
        self.metrics.all_trade_callbacks_received + 1
    if type(all_trade) ~= "table" then
        self.metrics.all_trade_dropped =
            self.metrics.all_trade_dropped + 1
        if should_log_counter(self.metrics.all_trade_dropped) then
            self.log("OnAllTrade dropped: invalid payload; received="
                .. tostring(self.metrics.all_trade_callbacks_received)
                .. "; dropped="
                .. tostring(self.metrics.all_trade_dropped), "warn")
        end
        return false
    end
    local class = normalize_class(all_trade.class_code)
    local security = trim(all_trade.sec_code)
    local record = self.books[key2(class, security)]
    local drop_reason = nil
    if not self.is_ready() then
        drop_reason = "bridge_not_ready"
    elseif not self:is_supported(class) then
        drop_reason = "unsupported_class"
    elseif not record then
        drop_reason = "no_dynamic_subscription"
    elseif not record.desired then
        drop_reason = "subscription_retiring"
    end
    if drop_reason then
        self.metrics.all_trade_dropped =
            self.metrics.all_trade_dropped + 1
        self.metrics.all_trade_unmatched =
            self.metrics.all_trade_unmatched + 1
        if should_log_counter(self.metrics.all_trade_unmatched) then
            self.log("OnAllTrade does not match an active CScalp book: "
                .. key2(class, security)
                .. "; reason=" .. tostring(drop_reason)
                .. "; trade_num=" .. tostring(all_trade.trade_num)
                .. "; unmatched="
                .. tostring(self.metrics.all_trade_unmatched)
                .. "; desired_books="
                .. tostring(#sorted_keys(self.books)), "info")
        end
        return false
    end

    record.all_trade_count = (record.all_trade_count or 0) + 1
    record.tick_all_trade_count =
        (record.tick_all_trade_count or 0) + 1
    record.last_all_trade_msec = self.now()
    record.last_all_trade_num = all_trade.trade_num
    if record.tick_handle then
        confirm_tick_source(self, record, record.tick_last_size,
            "OnAllTrade")
    end
    if suppress_historical_all_trade(self, record, all_trade) then
        return false
    end
    record.tick_forwarded_count =
        (record.tick_forwarded_count or 0) + 1
    if should_log_counter(record.all_trade_count) then
        self.log("OnAllTrade matched active subscription: " .. record.key
            .. "; count=" .. tostring(record.all_trade_count)
            .. "; generation_count="
            .. tostring(record.tick_all_trade_count)
            .. "; generation_forwarded="
            .. tostring(record.tick_forwarded_count)
            .. "; trade_num=" .. tostring(all_trade.trade_num)
            .. "; generation=" .. tostring(record.tick_generation)
            .. "; source_size=" .. tostring(record.tick_last_size)
            .. "; update_callbacks="
            .. tostring(record.tick_callback_count or 0), "info")
    end

    local compact = {}
    for i = 1, #TICK_FIELDS do
        local field = TICK_FIELDS[i]
        if all_trade[field] ~= nil then
            compact[field] = all_trade[field]
        end
    end
    local queued = enqueue_market_event(self, "OnAllTrade", compact)
    if not queued then
        self.metrics.all_trade_dropped =
            self.metrics.all_trade_dropped + 1
        return false
    end
    self.last_market_event_msec = self.now()
    self.metrics.all_trade_forwarded =
        self.metrics.all_trade_forwarded + 1
    return true
end

function M:on_param(class_code, sec_code)
    local class = normalize_class(class_code)
    local security = trim(sec_code)
    local key = key2(class, security)
    local record = self.books[key]
    if not self.is_ready() or not self:is_supported(class)
        or not record or not record.desired then
        self.metrics.param_dropped = self.metrics.param_dropped + 1
        return false
    end
    if self.pending_params[key] then
        self.metrics.param_coalesced = self.metrics.param_coalesced + 1
    end
    self.param_sequence = self.param_sequence + 1
    self.pending_params[key] = {
        key = key,
        class_code = class,
        sec_code = security,
        due_msec = self.now() + 100,
        sequence = self.param_sequence,
    }
    return true
end

function M:_server_time()
    local now = self.now()
    if self.server_time.value ~= nil
        and self.server_time.expires_msec > now then
        return self.server_time.value
    end
    local ok, value = pcall(self.api.get_info, "SERVERTIME")
    if ok then
        self.server_time.value = value
        self.server_time.expires_msec = now + 250
        return value
    end
    self.server_time.value = ""
    self.server_time.expires_msec = now + 250
    return ""
end

function M:get_quote(class_code, sec_code)
    local class = normalize_class(class_code)
    local security = trim(sec_code)
    local record = self.books[key2(class, security)]
    if not self.is_ready() then
        self.metrics.quote_dropped = self.metrics.quote_dropped + 1
        return nil, "bridge is not ready"
    end
    if not self:is_supported(class) or not record or not record.desired then
        self.metrics.quote_dropped = self.metrics.quote_dropped + 1
        return nil, "book is not subscribed"
    end
    local ok, quote = pcall(self.api.get_quote, class, security)
    if not ok then
        return nil, tostring(quote)
    end
    if type(quote) ~= "table" then
        return nil, "getQuoteLevel2 returned no table"
    end
    local result = shallow_copy(quote)
    result.class_code = class
    result.sec_code = security
    result.server_time = self:_server_time()
    self.last_market_event_msec = self.now()
    return result
end

function M:on_quote(class_code, sec_code)
    local quote, quote_error = self:get_quote(class_code, sec_code)
    if not quote then
        if quote_error ~= "book is not subscribed" then
            self.log("Quote callback failed for "
                .. key2(class_code, sec_code) .. ": "
                .. tostring(quote_error), "warn")
        end
        return false
    end
    local queued = enqueue_market_event(self, "OnQuote", quote)
    if not queued then
        self.metrics.quote_dropped = self.metrics.quote_dropped + 1
        return false
    end
    self.metrics.quote_forwarded = self.metrics.quote_forwarded + 1
    return true
end

local function fetch_candle(data_source, index)
    local methods = {
        low = "L",
        close = "C",
        high = "H",
        open = "O",
        volume = "V",
        datetime = "T",
    }
    local candle = {}
    for field, method_name in pairs(methods) do
        local method = data_source[method_name]
        if type(method) ~= "function" then
            return nil, "DataSource has no " .. method_name .. " method"
        end
        local ok, value = pcall(method, data_source, index)
        if not ok then
            return nil, method_name .. " failed: " .. tostring(value)
        end
        candle[field] = value
    end
    return candle
end

local function install_candle_callback(self, record, data_source, generation)
    if type(data_source.SetUpdateCallback) ~= "function" then
        return nil, "candle DataSource has no SetUpdateCallback method"
    end
    local callback = function(index)
        if record.generation ~= generation
            or record.handle ~= data_source or not record.desired then
            self.metrics.stale_generation_callbacks =
                self.metrics.stale_generation_callbacks + 1
            return
        end
        if not self.is_ready() then
            self.metrics.candle_dropped =
                self.metrics.candle_dropped + 1
            return
        end
        record.last_update_msec = self.now()
        local numeric_index = tonumber(index)
        if not numeric_index or numeric_index == record.last_index then
            return
        end
        record.last_index = numeric_index
        local completed_index = numeric_index - 1
        if completed_index < 1 then
            return
        end
        local candle, candle_error = fetch_candle(data_source, completed_index)
        if not candle then
            self.log("Candle callback read failed for " .. record.key
                .. ": " .. tostring(candle_error), "warn")
            return
        end
        candle.sec = record.sec_code
        candle.class = record.class_code
        candle.interval = record.interval
        local queued = enqueue_market_event(self, "NewCandle", candle)
        if not queued then
            self.metrics.candle_dropped =
                self.metrics.candle_dropped + 1
            return
        end
        self.last_market_event_msec = self.now()
    end
    local ok, result = pcall(
        data_source.SetUpdateCallback, data_source, callback)
    if not ok then
        return nil, tostring(result)
    end
    if result == false then
        return nil, "SetUpdateCallback returned false"
    end
    return true
end

local function create_candle_source(self, record)
    if not record.desired then
        return nil, "candle is no longer desired"
    end
    if not self.is_ready() then
        schedule_retry(self, record, "candle", "bridge is not ready")
        return nil, "bridge is not ready"
    end

    local ok, data_source, create_error = pcall(
        self.api.create_data_source,
        record.class_code,
        record.sec_code,
        record.interval)
    if not ok then
        data_source, create_error = nil, data_source
    end
    if not data_source or (create_error ~= nil and tostring(create_error) ~= "") then
        safe_close(self, data_source, "candle creation failed")
        local error_text = tostring(create_error or "CreateDataSource returned nil")
        schedule_retry(self, record, "candle", error_text)
        self.metrics.candle_create_failures =
            self.metrics.candle_create_failures + 1
        self.log("Candle subscription failed for " .. record.key
            .. "; retry_due_msec=" .. tostring(record.retry_due_msec)
            .. "; error=" .. tostring(error_text), "warn")
        return nil, error_text
    end

    local new_generation = (record.generation or 0) + 1
    local size, size_error = data_source_size(data_source)
    if size == nil then
        safe_close(self, data_source, "candle Size failed")
        schedule_retry(self, record, "candle", size_error)
        self.metrics.candle_create_failures =
            self.metrics.candle_create_failures + 1
        self.log("Candle Size failed for " .. record.key
            .. "; retry_due_msec=" .. tostring(record.retry_due_msec)
            .. "; error=" .. tostring(size_error), "warn")
        return nil, size_error
    end
    record.last_index = size
    local callback_ok, callback_error =
        install_candle_callback(self, record, data_source, new_generation)
    if not callback_ok then
        safe_close(self, data_source, "candle callback setup failed")
        schedule_retry(self, record, "candle", callback_error)
        self.metrics.candle_create_failures =
            self.metrics.candle_create_failures + 1
        self.log("Candle callback setup failed for " .. record.key
            .. "; retry_due_msec=" .. tostring(record.retry_due_msec)
            .. "; error=" .. tostring(callback_error), "warn")
        return nil, callback_error
    end

    local old_handle = record.handle
    record.handle = data_source
    record.generation = new_generation
    record.created_msec = self.now()
    record.last_update_msec = self.now()
    record.retry_attempt = 0
    record.retry_due_msec = nil
    record.last_error = nil
    record.retired_msec = nil
    safe_close(self, old_handle, "replaced candle " .. record.key)
    return true
end

function M:subscribe_candles(class_code, sec_code, interval, consumer_id)
    local class = normalize_class(class_code)
    local security = trim(sec_code)
    local numeric_interval = tonumber(interval)
    if not self:is_supported(class) then
        return nil, "unsupported class: " .. class
    end
    if security == "" or not numeric_interval or numeric_interval == 0 then
        return nil, "class, security and numeric interval are required"
    end
    local key = key3(class, security, numeric_interval)
    local record = self.candles[key]
    if not record then
        record = {
            key = key,
            class_code = class,
            sec_code = security,
            interval = numeric_interval,
            desired = true,
            generation = 0,
            retry_attempt = 0,
            owners = {},
        }
        self.candles[key] = record
    end
    record.owners = record.owners or {}
    record.owners[owner_key(consumer_id)] = true
    record.desired = true
    record.retired_msec = nil
    if record.handle then
        return true
    end
    return create_candle_source(self, record)
end

function M:unsubscribe_candles(class_code, sec_code, interval, consumer_id)
    local numeric_interval = tonumber(interval)
    if not numeric_interval then
        return nil, "numeric interval is required"
    end
    local key = key3(class_code, sec_code, numeric_interval)
    local record = self.candles[key]
    if not record then
        return true
    end
    record.owners = record.owners or {}
    record.owners[owner_key(consumer_id)] = nil
    if next(record.owners) ~= nil then
        return true
    end
    record.desired = false
    record.retry_due_msec = nil
    record.retired_msec = self.now()
    return true
end

function M:is_subscribed_candles(class_code, sec_code, interval, consumer_id)
    local numeric_interval = tonumber(interval)
    if not numeric_interval then
        return false
    end
    local record = self.candles[key3(class_code, sec_code, numeric_interval)]
    local owner = consumer_id ~= nil
        and owner_key(consumer_id) or nil
    return self.is_ready() and record ~= nil
        and record.desired and record.handle ~= nil
        and (owner == nil or record.owners[owner] == true)
end

function M:get_candles_snapshot(class_code, sec_code, interval, count,
    timeout_msec)
    local class = normalize_class(class_code)
    local security = trim(sec_code)
    local numeric_interval = tonumber(interval)
    local numeric_count = tonumber(count)
    if not self:is_supported(class) then
        return nil, "unsupported class: " .. class
    end
    if security == "" or not numeric_interval or numeric_interval == 0 then
        return nil, "class, security and non-zero numeric interval are required"
    end
    if not numeric_count or numeric_count < 0
        or numeric_count ~= math.floor(numeric_count) then
        return nil, "candle count must be a non-negative integer"
    end
    if not self.is_ready() then
        return nil, "bridge is not ready"
    end

    local key = key3(class, security, numeric_interval)
    local existing = self.candles[key]
    local data_source = existing and existing.desired
        and existing.handle or nil
    local temporary = false
    if not data_source then
        local ok, created, create_error = pcall(
            self.api.create_data_source, class, security, numeric_interval)
        if not ok then
            return nil, tostring(created)
        end
        if not created
            or (create_error ~= nil and tostring(create_error) ~= "") then
            safe_close(self, created,
                "temporary candle creation returned an error")
            return nil, tostring(create_error or "CreateDataSource returned nil")
        end
        data_source = created
        temporary = true
    end

    local timeout = math.max(1, tonumber(timeout_msec) or 7000)
    local deadline = self.now() + timeout
    local wait_attempts = 0
    local max_wait_attempts = math.ceil(timeout / 10) + 2
    local size, size_error = data_source_size(data_source)
    while size == 0 and self.now() < deadline
        and wait_attempts < max_wait_attempts do
        wait_attempts = wait_attempts + 1
        self.delay(10)
        self.pump()
        size, size_error = data_source_size(data_source)
        if size == nil then
            break
        end
    end
    if size == nil or size == 0 then
        if temporary then
            safe_close(self, data_source, "empty candle snapshot")
        end
        return nil, size_error or "timed out waiting for the first candle"
    end

    local start_index = numeric_count == 0 and 1
        or math.max(1, size - numeric_count + 1)
    local candles = {}
    for index = start_index, size do
        local candle, candle_error = fetch_candle(data_source, index)
        if not candle then
            if temporary then
                safe_close(self, data_source, "candle snapshot read failed")
            end
            return nil, candle_error
        end
        candle.sec = security
        candle.class = class
        candle.interval = numeric_interval
        candles[#candles + 1] = candle
    end
    if temporary then
        safe_close(self, data_source, "candle snapshot complete")
    end
    return candles
end

function M:mark_broker_disconnected()
    self.server_time.value = nil
    self.server_time.expires_msec = 0
    self.recovering = true
    self.metrics.param_dropped = self.metrics.param_dropped
        + #sorted_keys(self.pending_params)
    self.pending_params = {}
    discard_market_events(self, "broker disconnected")
    for _, record in pairs(self.books) do
        if record.desired then
            record.l2_active = false
            record.last_error = "broker disconnected"
        end
    end
    for _, record in pairs(self.candles) do
        if record.desired then
            record.last_error = "broker disconnected"
        end
    end
end

function M:begin_recovery()
    self.recovering = true
    self.metrics.param_dropped = self.metrics.param_dropped
        + #sorted_keys(self.pending_params)
    self.pending_params = {}
    discard_market_events(self, "broker recovery")
    self.recovery_books = {}
    self.recovery_candles = {}
    self.recovery_book_i = 1
    self.recovery_candle_i = 1

    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.desired then
            local closed = close_tick_source(self, record,
                "broker reconnect stale source")
            if closed then
                record.tick_handle = nil
                record.tick_state = nil
                record.tick_created_msec = nil
                record.tick_probe_due_msec = nil
                record.tick_diagnostic_due_msec = nil
                record.tick_diagnostic_count = 0
                record.tick_initial_size = nil
                record.tick_last_size = nil
                record.tick_callback_mode = nil
                record.tick_callback_count = 0
                record.tick_all_trade_count = 0
                record.tick_forwarded_count = 0
                record.tick_last_callback_index = nil
                record.tick_last_callback_msec = nil
                record.tick_first_callback_index = nil
                record.tick_first_callback_msec = nil
                record.tick_callback_logged_count = nil
                record.tick_callback_log_pending = false
                record.tick_callback_downgrade_pending = false
                record.tick_recreate_attempts = 0
                reset_tick_replay_state(record)
            else
                record.tick_state = "unconfirmed"
                record.tick_probe_due_msec = nil
                record.tick_diagnostic_due_msec =
                    self.now() + TICK_DIAGNOSTIC_FAST_MSEC
            end
            record.l2_active = false
            record.retry_due_msec = nil
            self.recovery_books[#self.recovery_books + 1] = key
        end
    end
    for _, key in ipairs(sorted_keys(self.candles)) do
        local record = self.candles[key]
        if record.desired then
            safe_close(self, record.handle,
                "broker reconnect stale candle " .. key)
            record.handle = nil
            record.retry_due_msec = nil
            self.recovery_candles[#self.recovery_candles + 1] = key
        end
    end
end

function M:recovery_step(limit)
    local budget = tonumber(limit) or 2
    local attempted = 0
    while attempted < budget
        and self.recovery_book_i <= #self.recovery_books do
        local key = self.recovery_books[self.recovery_book_i]
        self.recovery_book_i = self.recovery_book_i + 1
        local record = self.books[key]
        attempted = attempted + 1
        if record and record.desired then
            pcall(self.api.unsubscribe_l2, record.class_code, record.sec_code)
            activate_book(self, record)
        end
    end
    while attempted < budget
        and self.recovery_candle_i <= #self.recovery_candles do
        local key = self.recovery_candles[self.recovery_candle_i]
        self.recovery_candle_i = self.recovery_candle_i + 1
        local record = self.candles[key]
        attempted = attempted + 1
        if record and record.desired then
            create_candle_source(self, record)
        end
    end
    local now = self.now()
    if self.recovery_book_i > #self.recovery_books
        and self.recovery_candle_i > #self.recovery_candles then
        local book_keys = sorted_keys(self.books)
        for i = 1, #book_keys do
            if attempted >= budget then
                break
            end
            local record = self.books[book_keys[i]]
            if record.desired
                and (not record.l2_active or not record.tick_handle)
                and (not record.retry_due_msec
                    or record.retry_due_msec <= now) then
                activate_book(self, record)
                attempted = attempted + 1
            end
        end
        local candle_keys = sorted_keys(self.candles)
        for i = 1, #candle_keys do
            if attempted >= budget then
                break
            end
            local record = self.candles[candle_keys[i]]
            if record.desired and not record.handle
                and (not record.retry_due_msec
                    or record.retry_due_msec <= now) then
                create_candle_source(self, record)
                attempted = attempted + 1
            end
        end

        local unresolved = 0
        for _, record in pairs(self.books) do
            if record.desired
                and (not record.l2_active or not record.tick_handle) then
                unresolved = unresolved + 1
            end
        end
        for _, record in pairs(self.candles) do
            if record.desired and not record.handle then
                unresolved = unresolved + 1
            end
        end
        if unresolved == 0 then
            self.recovering = false
            return true
        end
    end
    return false
end

local function flush_params(self, now)
    if not self.is_ready() then
        return
    end
    local ready = {}
    for _, key in ipairs(sorted_keys(self.pending_params)) do
        local entry = self.pending_params[key]
        if entry and entry.due_msec <= now then
            ready[#ready + 1] = entry
        end
    end
    for i = 1, #ready do
        local entry = ready[i]
        local current = self.pending_params[entry.key]
        if current and current.sequence == entry.sequence then
            self.pending_params[entry.key] = nil
            self.metrics.param_forwarded =
                self.metrics.param_forwarded + 1
            self.last_market_event_msec = now
            self.emit("OnParam", {
                class_code = entry.class_code,
                sec_code = entry.sec_code,
            })
        end
    end
end

local function retry_one(self, now, records, creator)
    for _, key in ipairs(sorted_keys(records)) do
        local record = records[key]
        if record.desired and record.retry_due_msec
            and record.retry_due_msec <= now then
            creator(self, record)
            return true
        end
    end
    return false
end

local function audit_one_book(self, now)
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.desired and record.l2_active
            and (record.next_l2_audit_msec or 0) <= now then
            record.next_l2_audit_msec = now + 2000
            local ok, actual = pcall(
                self.api.is_subscribed_l2,
                record.class_code,
                record.sec_code)
            if ok and actual == false then
                record.l2_active = false
                schedule_retry(self, record, "tick",
                    "Level II subscription was lost")
                self.metrics.l2_audit_failures =
                    self.metrics.l2_audit_failures + 1
                self.log("Level II subscription audit failed for "
                    .. record.key .. "; repair scheduled", "warn")
            end
            return true
        end
    end
    return false
end

local function log_one_tick_callback_progress(self)
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.tick_callback_log_pending then
            record.tick_callback_log_pending = false
            local count = record.tick_callback_count or 0
            if record.tick_callback_logged_count == nil
                or should_log_counter(count) then
                local size, size_error = data_source_size(record.tick_handle)
                if size ~= nil then
                    record.tick_last_size = size
                end
                self.log("Tick DataSource update callback observed: "
                    .. record.key .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; callback_count=" .. tostring(count)
                    .. "; callback_index="
                    .. tostring(record.tick_last_callback_index)
                    .. "; source_size=" .. tostring(size)
                    .. "; size_error=" .. tostring(size_error)
                    .. "; OnAllTrade_count="
                    .. tostring(record.tick_all_trade_count or 0), "info")
                record.tick_callback_logged_count = count
            end
            return true
        end
    end
    return false
end

local function downgrade_one_tick_callback(self)
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.tick_callback_downgrade_pending
            and record.tick_handle then
            record.tick_callback_downgrade_pending = false
            local method = record.tick_handle.SetEmptyCallback
            if type(method) ~= "function" then
                self.metrics.tick_callback_downgrade_failures =
                    self.metrics.tick_callback_downgrade_failures + 1
                self.log("Tick callback cannot be downgraded: "
                    .. record.key .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; SetEmptyCallback is unavailable; lightweight "
                    .. "update callback remains active", "warn")
                return true
            end
            local ok, result = pcall(method, record.tick_handle)
            if ok and result ~= false then
                record.tick_callback_mode = "empty"
                self.metrics.tick_callback_downgrades =
                    self.metrics.tick_callback_downgrades + 1
                self.log("Tick diagnostic callback disabled after "
                    .. "OnAllTrade confirmation: " .. record.key
                    .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; callbacks_observed="
                    .. tostring(record.tick_callback_count or 0), "info")
                return true
            end
            self.metrics.tick_callback_downgrade_failures =
                self.metrics.tick_callback_downgrade_failures + 1
            self.log("Tick callback downgrade failed: " .. record.key
                .. "; generation=" .. tostring(record.tick_generation)
                .. "; error=" .. tostring(result)
                .. "; lightweight update callback remains active", "warn")
            return true
        end
    end
    return false
end

local function diagnose_one_tick_source(self, now)
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.desired and record.tick_handle
            and record.tick_state == "unconfirmed"
            and (record.tick_diagnostic_due_msec or 0) > 0
            and record.tick_diagnostic_due_msec <= now then
            local size, size_error = data_source_size(record.tick_handle)
            if size ~= nil then
                record.tick_last_size = size
            end
            local native = native_all_trades_snapshot(self)
            record.tick_diagnostic_count =
                (record.tick_diagnostic_count or 0) + 1
            self.metrics.tick_diagnostic_snapshots =
                self.metrics.tick_diagnostic_snapshots + 1

            local classification = "source_state_ambiguous"
            if size == nil then
                classification = "datasource_size_failed"
            elseif (record.tick_callback_count or 0) > 0
                and (record.tick_all_trade_count or 0) == 0 then
                classification = "datasource_active_OnAllTrade_missing"
            elseif size > (record.tick_initial_size or 0)
                and (record.tick_all_trade_count or 0) == 0 then
                classification = "datasource_growing_OnAllTrade_missing"
            elseif size == 0 and (record.tick_callback_count or 0) == 0
                and native.count == 0 then
                classification = "native_all_trades_stream_empty"
            elseif size == 0 and (record.tick_callback_count or 0) == 0
                and native.count and native.count > 0 then
                classification = "native_table_active_requested_source_idle"
            end

            self.log("Tick subscription diagnostic: " .. record.key
                .. "; classification=" .. classification
                .. "; sample="
                .. tostring(record.tick_diagnostic_count)
                .. "; generation=" .. tostring(record.tick_generation)
                .. "; state=" .. tostring(record.tick_state)
                .. "; callback_mode="
                .. tostring(record.tick_callback_mode)
                .. "; update_callbacks="
                .. tostring(record.tick_callback_count or 0)
                .. "; last_callback_index="
                .. tostring(record.tick_last_callback_index)
                .. "; source_initial_size="
                .. tostring(record.tick_initial_size)
                .. "; source_size=" .. tostring(size)
                .. "; source_size_error=" .. tostring(size_error)
                .. "; OnAllTrade_count="
                .. tostring(record.tick_all_trade_count or 0)
                .. "; owners=" .. tostring(owner_count(record))
                .. "; age_msec="
                .. tostring(now - (record.tick_created_msec or now))
                .. "; " .. native_snapshot_text(native), "info")

            if (record.tick_callback_count or 0) > 0
                and record.tick_callback_mode == "update" then
                record.tick_callback_downgrade_pending = true
            end

            local delay = record.tick_diagnostic_count
                < TICK_DIAGNOSTIC_FAST_LIMIT
                and TICK_DIAGNOSTIC_FAST_MSEC
                or TICK_DIAGNOSTIC_SLOW_MSEC
            record.tick_diagnostic_due_msec = now + delay
            return true
        end
    end
    return false
end

local function audit_one_tick_source(self, now)
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        if record.desired and record.l2_active and record.tick_handle
            and record.tick_state == "warming"
            and (record.tick_probe_due_msec or 0) <= now then
            local size, size_error = data_source_size(record.tick_handle)
            if size == nil then
                record.tick_state = "unconfirmed"
                record.tick_probe_due_msec = nil
                record.tick_diagnostic_due_msec =
                    now + TICK_DIAGNOSTIC_FAST_MSEC
                self.metrics.tick_probe_failures =
                    self.metrics.tick_probe_failures + 1
                self.log("Tick subscription audit failed: " .. record.key
                    .. "; generation=" .. tostring(record.tick_generation)
                    .. "; error=" .. tostring(size_error)
                    .. "; " .. native_snapshot_text(
                        native_all_trades_snapshot(self)), "warn")
                return true
            end
            record.tick_last_size = size
            local initial_size = record.tick_initial_size or 0
            if initial_size > 0 or size > initial_size
                or (record.tick_callback_count or 0) > 0 then
                record.tick_state = "unconfirmed"
                record.tick_probe_due_msec = nil
                record.tick_diagnostic_due_msec =
                    now + TICK_DIAGNOSTIC_FAST_MSEC
                if record.tick_callback_mode == "update" then
                    record.tick_callback_downgrade_pending = true
                end
                self.log("Tick DataSource is receiving data but OnAllTrade "
                    .. "confirmation is pending: " .. record.key
                    .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; initial_size=" .. tostring(initial_size)
                    .. "; current_size=" .. tostring(size)
                    .. "; update_callbacks="
                    .. tostring(record.tick_callback_count or 0)
                    .. "; source is retained", "warn")
                return true
            end

            local recreates = record.tick_recreate_attempts or 0
            if recreates >= TICK_STARTUP_RECREATE_LIMIT then
                record.tick_state = "unconfirmed"
                record.tick_probe_due_msec = nil
                record.tick_diagnostic_due_msec =
                    now + TICK_DIAGNOSTIC_FAST_MSEC
                self.log("Tick subscription remains unconfirmed: "
                    .. record.key .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; initial_size="
                    .. tostring(initial_size)
                    .. "; current_size=" .. tostring(size)
                    .. "; recreates=" .. tostring(recreates)
                    .. "; automatic recreation limit reached; "
                    .. native_snapshot_text(
                        native_all_trades_snapshot(self)), "warn")
                return true
            end

            self.metrics.tick_startup_stalls =
                self.metrics.tick_startup_stalls + 1
            local closed = close_tick_source(self, record,
                "stalled tick startup")
            if not closed then
                record.tick_state = "unconfirmed"
                record.tick_probe_due_msec = nil
                self.metrics.tick_recreate_failures =
                    self.metrics.tick_recreate_failures + 1
                self.log("Tick subscription recreation aborted: "
                    .. record.key .. "; generation="
                    .. tostring(record.tick_generation)
                    .. "; old DataSource could not be closed", "error")
                return true
            end
            record.tick_handle = nil
            record.tick_state = "restarting"
            record.tick_created_msec = nil
            record.tick_probe_due_msec = nil
            record.tick_diagnostic_due_msec = nil
            record.tick_diagnostic_count = 0
            record.tick_initial_size = nil
            record.tick_last_size = nil
            record.tick_callback_mode = nil
            record.tick_callback_count = 0
            record.tick_all_trade_count = 0
            record.tick_forwarded_count = 0
            record.tick_last_callback_index = nil
            record.tick_last_callback_msec = nil
            record.tick_first_callback_index = nil
            record.tick_first_callback_msec = nil
            record.tick_callback_logged_count = nil
            record.tick_callback_log_pending = false
            record.tick_callback_downgrade_pending = false
            record.tick_recreate_attempts = recreates + 1
            reset_tick_replay_state(record)
            self.metrics.tick_recreates =
                self.metrics.tick_recreates + 1
            schedule_retry(self, record, "tick",
                "tick DataSource stayed empty during startup")
            self.log("Tick subscription stalled: " .. record.key
                .. "; recreation="
                .. tostring(record.tick_recreate_attempts)
                .. "/" .. tostring(TICK_STARTUP_RECREATE_LIMIT)
                .. "; initial_size="
                .. tostring(initial_size)
                .. "; current_size=" .. tostring(size)
                .. "; retry_due_msec="
                .. tostring(record.retry_due_msec), "warn")
            return true
        end
    end
    return false
end

local function retire_sources(self, now)
    local book_keys = sorted_keys(self.books)
    for i = 1, #book_keys do
        local key = book_keys[i]
        local record = self.books[key]
        if not record.desired and record.retired_msec then
            local age = now - record.retired_msec
            if age >= 1500
                and (not another_desired_book(self, key)
                    or another_active_book(self, key)
                    or age >= 10000) then
                if close_book_record(self, record, "retired") then
                    self.books[key] = nil
                else
                    record.retired_msec = now
                    self.log("Retired market resource will be retried: "
                        .. key, "warn")
                end
            end
        end
    end

    local candle_keys = sorted_keys(self.candles)
    for i = 1, #candle_keys do
        local key = candle_keys[i]
        local record = self.candles[key]
        if not record.desired and record.retired_msec
            and now - record.retired_msec >= 1500 then
            safe_close(self, record.handle, "retired candle " .. key)
            self.candles[key] = nil
        end
    end
end

function M:housekeeping()
    local now = self.now()
    flush_market_events(self)
    flush_params(self, now)
    log_one_tick_callback_progress(self)
    downgrade_one_tick_callback(self)
    if not self.recovering and self.is_ready() then
        if now - self.last_l2_audit_msec >= 500 then
            self.last_l2_audit_msec = now
            audit_one_book(self, now)
            audit_one_tick_source(self, now)
            diagnose_one_tick_source(self, now)
        end
        if not retry_one(self, now, self.books, activate_book) then
            retry_one(self, now, self.candles, create_candle_source)
        end
    end
    retire_sources(self, now)
end

function M:cleanup(reason)
    local cleanup_reason = tostring(reason or "cleanup")
    local retained_books = {}
    local released_books = 0
    local failed_books = 0
    local book_keys = sorted_keys(self.books)
    for i = 1, #book_keys do
        local key = book_keys[i]
        local record = self.books[key]
        record.desired = false
        record.owners = {}
        record.retry_due_msec = nil
        if close_book_record(self, record, cleanup_reason) then
            released_books = released_books + 1
        else
            failed_books = failed_books + 1
            record.retired_msec = self.now()
            retained_books[key] = record
        end
    end
    local candle_keys = sorted_keys(self.candles)
    for i = 1, #candle_keys do
        local record = self.candles[candle_keys[i]]
        safe_close(self, record.handle,
            cleanup_reason .. " candle " .. record.key)
    end
    self.books = retained_books
    self.candles = {}
    self.pending_params = {}
    discard_market_events(self, reason or "cleanup")
    self.recovery_books = {}
    self.recovery_candles = {}
    self.recovering = false
    self.log("Market cleanup completed; reason=" .. cleanup_reason
        .. "; released_books=" .. tostring(released_books)
        .. "; retained_after_close_failure=" .. tostring(failed_books)
        .. "; candles=" .. tostring(#candle_keys),
        failed_books > 0 and "warn" or "info")
end

function M:health()
    local desired_books = 0
    local book_consumers = 0
    local active_books = 0
    local tick_sources = 0
    local confirmed_tick_sources = 0
    local warming_tick_sources = 0
    local unconfirmed_tick_sources = 0
    local desired_candles = 0
    local candle_consumers = 0
    local candle_sources = 0
    local pending_retries = 0
    local tick_subscription_details = {}
    for _, record in pairs(self.books) do
        if record.desired then
            desired_books = desired_books + 1
            book_consumers = book_consumers + owner_count(record)
        end
        if record.l2_active then
            active_books = active_books + 1
        end
        if record.tick_handle then
            tick_sources = tick_sources + 1
            if record.tick_state == "confirmed" then
                confirmed_tick_sources = confirmed_tick_sources + 1
            elseif record.tick_state == "warming" then
                warming_tick_sources = warming_tick_sources + 1
            else
                unconfirmed_tick_sources =
                    unconfirmed_tick_sources + 1
            end
        end
        if record.retry_due_msec then
            pending_retries = pending_retries + 1
        end
    end
    for _, record in pairs(self.candles) do
        if record.desired then
            desired_candles = desired_candles + 1
            candle_consumers = candle_consumers + owner_count(record)
        end
        if record.handle then
            candle_sources = candle_sources + 1
        end
        if record.retry_due_msec then
            pending_retries = pending_retries + 1
        end
    end
    for _, key in ipairs(sorted_keys(self.books)) do
        local record = self.books[key]
        tick_subscription_details[#tick_subscription_details + 1] = {
            key = record.key,
            desired = record.desired == true,
            owners = owner_count(record),
            l2_active = record.l2_active == true,
            tick_handle = record.tick_handle ~= nil,
            tick_state = record.tick_state,
            tick_generation = record.tick_generation,
            callback_mode = record.tick_callback_mode,
            update_callbacks = record.tick_callback_count or 0,
            last_callback_index = record.tick_last_callback_index,
            source_initial_size = record.tick_initial_size,
            source_last_size = record.tick_last_size,
            all_trade_count = record.all_trade_count or 0,
            generation_all_trade_count =
                record.tick_all_trade_count or 0,
            generation_forwarded_count =
                record.tick_forwarded_count or 0,
            replay_filter_enabled =
                record.tick_replay_filter_enabled == true,
            replay_suppressed = record.tick_replay_dropped or 0,
            replay_caught_up = record.tick_replay_caught_up == true,
            replay_clock_error = record.tick_replay_clock_error,
            last_all_trade_num = record.last_all_trade_num,
            diagnostic_count = record.tick_diagnostic_count or 0,
            last_error = record.last_error,
        }
    end
    return {
        desired_books = desired_books,
        book_consumers = book_consumers,
        active_books = active_books,
        tick_sources = tick_sources,
        confirmed_tick_sources = confirmed_tick_sources,
        warming_tick_sources = warming_tick_sources,
        unconfirmed_tick_sources = unconfirmed_tick_sources,
        desired_candles = desired_candles,
        candle_consumers = candle_consumers,
        candle_sources = candle_sources,
        pending_retries = pending_retries,
        pending_params = #sorted_keys(self.pending_params),
        market_event_queue_depth = market_event_depth(self),
        recovering = self.recovering,
        last_market_event_msec = self.last_market_event_msec,
        native_all_trades = shallow_copy(self.last_native_all_trades),
        tick_subscription_details = tick_subscription_details,
        metrics = shallow_copy(self.metrics),
    }
end

return M