⚡️ Speed up method lbank.fetch_deposits by 40%

[codeflash-ai]

📄 40% (0.40x) speedup for lbank.fetch_deposits in python/ccxt/async_support/lbank.py

⏱️ Runtime : 296 microseconds → 212 microseconds (best of 97 runs)

📝 Explanation and details

The optimized code achieves a 39% runtime improvement and 1.0% throughput improvement through three key micro-optimizations in the base Exchange class methods that are heavily called during transaction processing:

Key Optimizations:

1. safe_value() fast path: Added a direct dict.get() call for the common case of dict + str key combinations, avoiding the expensive Exchange.key_exists() function call. Line profiler shows this reduced execution time from 219ms to 98ms (55% improvement).

2. extend() method streamlining: Eliminated redundant type checks and variable assignments, using cleaner conditional logic with early empty-args handling. The optimization reduces overhead by ~30ms while maintaining OrderedDict compatibility.

3. safe_list() fast path: Similar to safe_value, this adds a direct path for dict/str combinations that immediately checks if the value is a list, avoiding the slower safe_list_n() fallback. Execution time dropped from 250ms to 91ms (64% improvement).

Why These Optimizations Matter:

From the annotated tests, fetch_deposits processes financial transaction data where these utility methods are called repeatedly - safe_value is hit 103 times and safe_list 73 times per call. The optimizations are particularly effective for:

• Large deposit lists (250+ records) where parsing overhead compounds
• High-frequency API calls in trading applications
• Bulk transaction processing scenarios

The optimizations maintain full backward compatibility while providing substantial performance gains for the most common usage patterns (dict/string key combinations), making them ideal for production cryptocurrency exchange integrations where every microsecond counts.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 104 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import asyncio

import pytest
from ccxt.async_support.lbank import lbank

# Import the lbank class and its dependencies as defined above.
# For the sake of this test file, we'll assume that all dependencies and parent classes are available in the test environment.

# ----------------- TEST SUITE FOR lbank.fetch_deposits -----------------

@pytest.fixture
def lbank_instance(monkeypatch):
    """
    Returns an instance of lbank with required methods stubbed for isolated testing.
    We stub only the lowest-level network methods (spotPrivatePostSupplementDepositHistory, currency, parse_transactions)
    so the rest of the logic is exercised as in production.
    """
    # Create an instance
    lb = lbank({})

    # Patch load_markets to be a no-op (it is always awaited, but doesn't need to do anything for our tests)
    async def load_markets_stub(*args, **kwargs):
        return None
    lb.load_markets = load_markets_stub

    # Patch currency to return a dummy currency mapping for code lookups
    def currency_stub(code):
        # Simulate a mapping from code to currency dict
        mapping = {
            'USDT': {'id': 'usdt', 'code': 'USDT'},
            'BTC': {'id': 'btc', 'code': 'BTC'},
            'ETH': {'id': 'eth', 'code': 'ETH'},
        }
        if code in mapping:
            return mapping[code]
        raise Exception(f"Currency code {code} not found")
    lb.currency = currency_stub

    # Patch parse_transactions to just return the input for easy checking
    def parse_transactions_stub(deposits, currency, since, limit):
        # For test purposes, just return the deposits list, optionally sliced by limit
        if limit is not None:
            return deposits[:limit]
        return deposits
    lb.parse_transactions = parse_transactions_stub

    return lb

# ----------------- BASIC TEST CASES -----------------

@pytest.mark.asyncio
async def test_fetch_deposits_returns_empty_list_when_no_deposits(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits returns an empty list when there are no deposits.
    """
    # Patch the HTTP call to return empty depositOrders
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "data": {
                "total": 0,
                "depositOrders": [],
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits()

@pytest.mark.asyncio

async def test_fetch_deposits_with_code_param(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits correctly passes the 'coin' parameter when code is provided.
    """
    # We'll check that the request dict includes the correct coin id
    called = {}

    async def spotPrivatePostSupplementDepositHistory_stub(request):
        called['request'] = request.copy()
        return {
            "result": True,
            "data": {
                "total": 1,
                "depositOrders": [{"txId": "TXID3", "coin": "btc"}],
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits(code="BTC")

@pytest.mark.asyncio
async def test_fetch_deposits_with_since_param(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits correctly passes the 'startTime' parameter when since is provided.
    """
    called = {}

    async def spotPrivatePostSupplementDepositHistory_stub(request):
        called['request'] = request.copy()
        return {
            "result": True,
            "data": {
                "total": 1,
                "depositOrders": [{"txId": "TXID4", "coin": "eth"}],
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits(code="ETH", since=1650000000000)

@pytest.mark.asyncio
async def test_fetch_deposits_with_limit_param(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits applies the limit argument to the returned list.
    """
    deposit_orders = [
        {"txId": "TX1", "coin": "usdt"},
        {"txId": "TX2", "coin": "usdt"},
        {"txId": "TX3", "coin": "usdt"},
    ]
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "data": {
                "total": 3,
                "depositOrders": deposit_orders,
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits(limit=2)

# ----------------- EDGE TEST CASES -----------------

@pytest.mark.asyncio

async def test_fetch_deposits_handles_missing_data_key(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits returns an empty list if the 'data' key is missing in response.
    """
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits()

@pytest.mark.asyncio
async def test_fetch_deposits_handles_missing_depositOrders_key(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits returns an empty list if 'depositOrders' is missing in 'data'.
    """
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "data": {
                "total": 0,
                # 'depositOrders' key is missing
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    result = await lbank_instance.fetch_deposits()

@pytest.mark.asyncio

async def test_fetch_deposits_handles_non_list_depositOrders(lbank_instance, monkeypatch):
    """
    Test that fetch_deposits handles 'depositOrders' being a dict (should convert to list).
    """
    deposit_orders_dict = {
        "0": {"txId": "TXDICT1", "coin": "usdt"},
        "1": {"txId": "TXDICT2", "coin": "usdt"},
    }
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "data": {
                "total": 2,
                "depositOrders": deposit_orders_dict,
                "page_length": 20,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    # The parse_transactions stub just returns the input as a list
    result = await lbank_instance.fetch_deposits()

# ----------------- LARGE SCALE TEST CASES -----------------

@pytest.mark.asyncio




async def test_fetch_deposits_throughput_high_volume(lbank_instance, monkeypatch):
    """
    Throughput test: high concurrency with large deposit lists.
    """
    deposit_orders = [{"txId": f"TX{i}", "coin": "usdt"} for i in range(250)]
    async def spotPrivatePostSupplementDepositHistory_stub(request):
        return {
            "result": True,
            "data": {
                "total": 250,
                "depositOrders": deposit_orders,
                "page_length": 250,
                "current_page": 1
            },
            "error_code": 0,
            "ts": 1649719721758
        }
    lbank_instance.spotPrivatePostSupplementDepositHistory = spotPrivatePostSupplementDepositHistory_stub

    # Simulate 30 concurrent requests (total 30*250 = 7500 records processed)
    coros = [lbank_instance.fetch_deposits() for _ in range(30)]
    results = await asyncio.gather(*coros)
    for result in results:
        pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import asyncio

import pytest
from ccxt.async_support.lbank import lbank
from ccxt.base.types import Transaction

# ---------------------------
# Basic Test Cases
# ---------------------------

@pytest.mark.asyncio

To edit these changes git checkout codeflash/optimize-lbank.fetch_deposits-mhxo3gh8 and push.