成交量最高

Most Traded

专题: Algorithmic Programming / 算法编程
难度: L2
来源: OpenQuant

题目详情

给定一条股票数据流，设计一个数据结构，高效支持以下操作：

`execute_trade(ticker, volume)` - 记录一笔已经发生的成交
`most_traded(k)` - 返回按成交量排序的前 $k$ 只股票。返回格式应为字符串列表，每个字符串格式如下：` `

示例：

st = StockTracker()
st.execute_trade('TSLA', 1000)
st.execute_trade('NFLX', 700)
st.execute_trade('TSLA', 200)
st.execute_trade('META', 1400)
st.most_traded(2)
# This should return the following:
# META 1400
# TSLA 1200

请完成下面的类定义：

from typing import List

class StockTracker:
    def __init__(self) -> None:
        pass

    def execute_trade(self, ticker: str, volume: int) -> None:
        pass

    def most_traded(self, k: int) -> List[str]:
        pass

Given a stream of stock data, create a data structure that can efficiently handle the following functions:

`execute_trade(ticker, volume)` - store the trade that has occurred
`most_traded(k)` - return the $k$ most traded stocks by volume. The return format should be a list of strings with each string being formatted as: ` `

Example:

st = StockTracker()
st.execute_trade('TSLA', 1000)
st.execute_trade('NFLX', 700)
st.execute_trade('TSLA', 200)
st.execute_trade('META', 1400)
st.most_traded(2)
# This should return the following:
# META 1400
# TSLA 1200

Complete the following class for your solution

from typing import List

class StockTracker:
    def __init__(self) -> None:
        pass

    def execute_trade(self, ticker: str, volume: int) -> None:
        pass

    def most_traded(self, k: int) -> List[str]:
        pass

解析

可以用 `hash map + 大小为 k 的最小堆` 来做。

思路如下：

`execute_trade(ticker, volume)`：用哈希表累计每个 `ticker` 的总成交量。
`most_traded(k)`：遍历哈希表，把 `(volume, ticker)` 放入最小堆；当堆大小超过 `k` 时弹出最小元素，这样最后堆里保留的就是成交量最高的前 `k` 只股票。

复杂度：

`execute_trade`：时间 $O(1)$ ，空间由哈希表决定。
`most_traded`：若共有 $n$ 个股票代码，则时间 $O(n \log k)$ ，额外空间 $O(k)$ 。

import heapq
from typing import List

class StockTracker:
    def __init__(self) -> None:
        self.trade_volume = {}

    def execute_trade(self, ticker: str, volume: int) -> None:
        self.trade_volume[ticker] = self.trade_volume.get(ticker, 0) + volume

    def most_traded(self, k: int) -> List[str]:
        min_heap = []

        for ticker, volume in self.trade_volume.items():
            heapq.heappush(min_heap, (volume, ticker))
            if len(min_heap) > k:
                heapq.heappop(min_heap)

        top_k = sorted(min_heap, reverse=True)
        return [f"{ticker} {volume}" for volume, ticker in top_k]

Original Explanation

To solve this question we can leverage a hashmap and a min heap. The hashmap will be used to store the ticker and corresponding trade volume for each trade that is executed. The min heap will be used to retrieve the most traded stocks by volume. When the `most_traded` function is called, we will populate the heap with the $n$ most traded stocks from our hashmap.

Solution:

`execute_trade`
- Time Complexity: $O(1)$ - inserting into a map is constant time
- Space Complexity: $O(n)$ - we can have up to $n$ tickers in the map
`most_traded`
- Time Complexity: $O(n \log k)$ - we iterate over the $n$ tickers in the map and each push operation takes $\log(k)$ time, where $k$ is the number of elements in the heap at any given point in time.
- Space Complexity: $O(k)$ - there can be at most $k$ elements in the heap at any point in time.

import heapq
from typing import List

class StockTracker:
    def __init__(self) -> None:
        self.trade_volume = {}

    def execute_trade(self, ticker: str, volume: int) -> None:
        # Set the ticker trade volume to the current volume + new trade volume
        self.trade_volume[ticker] = self.trade_volume.get(ticker, 0) + volume

    def most_traded(self, k: int) -> List[str]:
        min_heap = []

        for ticker, volume in self.trade_volume.items():
            # Add the ticker and volume to the min heap
            # The heap will sort by the first value in the tuple
            heapq.heappush(min_heap, (volume, ticker))

            # We only want to top n most traded stocks so we remove any extras
            if len(max_heap) > k:
                # Removes the smallest value from the heap
                heapq.heappop(min_heap)

        # Format the output
        output = [f"{ticker} {volume}" for volume, ticker in max_heap]

        # Reverse the output so that it is by volume in descending order
        return output[::-1]