1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

def heap_sort(arr: list) -> None:
    """
    Sort an array in-place using a MAX-heap.
    
    The algorithm:
    1. Build a max-heap from the array (bottom-up heapify)
    2. Repeatedly:
       a. Swap root (maximum) with last element
       b. Reduce heap size by 1
       c. Heapify down from root to restore heap property
    
    Result: Array is sorted in ASCENDING order
    
    Time Complexity: O(n log n)
    Space Complexity: O(1) - in-place
    """
    n = len(arr)
    
    # Step 1: Build max-heap (bottom-up)
    # Start from last non-leaf and heapify each node
    for i in range(n // 2 - 1, -1, -1):
        _heapify_down(arr, n, i)
    
    # Step 2: Extract maximums one by one
    for i in range(n - 1, 0, -1):
        # Swap maximum (root) with current last element
        arr[0], arr[i] = arr[i], arr[0]
        # Heapify the reduced heap (size = i, not n)
        _heapify_down(arr, i, 0)
 
def _heapify_down(arr: list, heap_size: int, root_idx: int) -> None:
    """
    Restore max-heap property starting from root_idx.
    
    The parent should be LARGER than both children.
    If violated, swap with larger child and recurse.
    """
    largest = root_idx
    left = 2 * root_idx + 1
    right = 2 * root_idx + 2
    
    # Check if left child exists and is larger than root
    if left < heap_size and arr[left] > arr[largest]:
        largest = left
    
    # Check if right child exists and is larger than current largest
    if right < heap_size and arr[right] > arr[largest]:
        largest = right
    
    # If largest is not root, swap and continue heapifying
    if largest != root_idx:
        arr[root_idx], arr[largest] = arr[largest], arr[root_idx]
        _heapify_down(arr, heap_size, largest)
 
# Example usage
data = [12, 11, 13, 5, 6, 7]
heap_sort(data)
print(data)  # [5, 6, 7, 11, 12, 13] - ascending order

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

import heapq
from typing import List
 
def k_smallest_elements(arr: List[int], k: int) -> List[int]:
    """
    Find the K smallest elements using a MAX-heap.
    
    Since Python's heapq is a min-heap, we negate values to simulate max-heap.
    This is a common pattern in Python for max-heap behavior.
    
    The algorithm maintains a "ceiling":
    - The max-heap root is the K-th smallest seen so far (the largest of the K smallest)
    - Any element larger than this ceiling cannot be in the K smallest
    - Any element smaller than this ceiling replaces the current ceiling
    
    Time Complexity: O(n log k) - process n elements, each heap op is O(log k)
    Space Complexity: O(k) - only store k elements
    """
    if k <= 0:
        return []
    if k >= len(arr):
        return sorted(arr)
    
    # Build max-heap of first k elements (negate for max-heap in Python)
    max_heap = [-x for x in arr[:k]]
    heapq.heapify(max_heap)
    
    # Process remaining elements
    for i in range(k, len(arr)):
        current = arr[i]
        # -max_heap[0] is the current maximum (K-th smallest)
        if current < -max_heap[0]:
            # New element is smaller than current K-th smallest
            # Replace the K-th smallest with this element
            heapq.heapreplace(max_heap, -current)
    
    # Return the K smallest (un-negate and sort)
    return sorted(-x for x in max_heap)
 
def k_smallest_elements_with_indices(arr: List[int], k: int) -> List[tuple]:
    """
    Find K smallest elements along with their original indices.
    Demonstrates max-heap with complex items.
    """
    if k <= 0:
        return []
    if k >= len(arr):
        return sorted(enumerate(arr), key=lambda x: x[1])
    
    # Max-heap entries: (-value, index)
    # Negating value gives max-heap behavior
    max_heap = [(-arr[i], i) for i in range(k)]
    heapq.heapify(max_heap)
    
    for i in range(k, len(arr)):
        if arr[i] < -max_heap[0][0]:
            heapq.heapreplace(max_heap, (-arr[i], i))
    
    # Return (index, value) pairs sorted by value
    return sorted([(idx, -val) for val, idx in max_heap], key=lambda x: x[1])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

import heapq
from typing import List, Tuple
 
def job_scheduling_max_profit(jobs: List[Tuple[int, int, int]]) -> int:
    """
    Maximize profit from job scheduling.
    Each job: (deadline, profit, job_id)
    Each job takes 1 unit of time. We can do at most one job per time slot.
    
    Greedy approach: 
    - Sort jobs by deadline
    - Use max-heap to always have access to highest profit job
    - For each time slot, pick the highest profit job available
    
    Time Complexity: O(n log n)
    """
    if not jobs:
        return 0
    
    # Sort by deadline (ascending)
    jobs_sorted = sorted(jobs, key=lambda x: x[0])
    
    # Max-heap of (profit, job_id) for jobs that could be scheduled
    # Negate profit for max-heap behavior in Python
    max_heap = []
    
    total_profit = 0
    current_time = 0
    job_idx = 0
    n = len(jobs_sorted)
    
    # Find the maximum deadline to know how many time slots exist
    max_deadline = max(job[0] for job in jobs_sorted)
    
    # Process each time slot from 1 to max_deadline
    for time_slot in range(1, max_deadline + 1):
        # Add all jobs with deadline >= current time slot to the heap
        while job_idx < n and jobs_sorted[job_idx][0] <= time_slot:
            deadline, profit, job_id = jobs_sorted[job_idx]
            heapq.heappush(max_heap, (-profit, job_id))  # Negate for max-heap
            job_idx += 1
        
        # Pick the highest profit job available for this time slot
        if max_heap:
            neg_profit, job_id = heapq.heappop(max_heap)
            total_profit += -neg_profit
    
    return total_profit
 
# Alternative: Classic Job Sequencing Problem
def job_sequencing_max_profit(jobs: List[Tuple[int, int]]) -> Tuple[int, List[int]]:
    """
    Classic job sequencing: each job has (deadline, profit).
    Maximize total profit. Return total profit and sequence.
    
    Approach: Greedy with max-heap
    - Sort jobs by deadline descending
    - For each job, try to schedule at its deadline or earlier
    """
    if not jobs:
        return 0, []
    
    # Sort by deadline descending
    n = len(jobs)
    indexed_jobs = [(deadline, profit, i) for i, (deadline, profit) in enumerate(jobs)]
    indexed_jobs.sort(key=lambda x: x[0], reverse=True)
    
    # Track which time slots are used
    max_deadline = max(job[0] for job in indexed_jobs)
    slot_used = [False] * (max_deadline + 1)
    
    # Max-heap by profit
    max_heap = []
    
    result_sequence = []
    total_profit = 0
    job_idx = 0
    
    # Process time slots from max_deadline down to 1
    for slot in range(max_deadline, 0, -1):
        # Add all jobs with deadline >= slot to heap
        while job_idx < n and indexed_jobs[job_idx][0] >= slot:
            deadline, profit, original_idx = indexed_jobs[job_idx]
            heapq.heappush(max_heap, (-profit, original_idx))
            job_idx += 1
        
        # Schedule highest profit job for this slot
        if max_heap:
            neg_profit, original_idx = heapq.heappop(max_heap)
            total_profit += -neg_profit
            result_sequence.append(original_idx)
    
    return total_profit, result_sequence[::-1]  # Reverse for chronological order

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

import heapq
from typing import List
 
def sliding_window_maximum(nums: List[int], k: int) -> List[int]:
    """
    Find maximum in each sliding window of size k using a max-heap.
    
    Key insight: Lazy deletion
    - We don't remove elements when they leave the window immediately
    - Instead, we check if the maximum's index is within the current window
    - If not, we pop and check the next maximum
    
    This works because:
    - An element outside the window can never be the answer
    - We only care about the maximum, so invalid elements below it don't matter
    
    Time Complexity: O(n log n) worst case
    Space Complexity: O(n) for the heap
    """
    if not nums or k <= 0:
        return []
    
    n = len(nums)
    result = []
    
    # Max-heap: (-value, index) - negate for max-heap behavior
    max_heap = []
    
    for i in range(n):
        # Add current element to heap
        heapq.heappush(max_heap, (-nums[i], i))
        
        # Once we have a full window
        if i >= k - 1:
            # Remove elements outside the current window
            # (Lazy deletion: only remove when they reach the top)
            while max_heap[0][1] <= i - k:
                heapq.heappop(max_heap)
            
            # The root now contains the valid maximum
            result.append(-max_heap[0][0])
    
    return result
 
def running_maximum_with_removal(operations: List[tuple]) -> List[int]:
    """
    Support insert, delete, and query maximum operations.
    
    Operations: ('insert', value), ('delete', value), ('max',)
    
    Max-heap with lazy deletion for deleted elements.
    """
    max_heap = []
    deleted = {}  # Count of deleted values not yet removed from heap
    results = []
    
    for op in operations:
        if op[0] == 'insert':
            value = op[1]
            heapq.heappush(max_heap, -value)
        
        elif op[0] == 'delete':
            value = op[1]
            # Mark as deleted (lazy)
            deleted[value] = deleted.get(value, 0) + 1
        
        elif op[0] == 'max':
            # Clean up deleted elements at root
            while max_heap and deleted.get(-max_heap[0], 0) > 0:
                val = -heapq.heappop(max_heap)
                deleted[val] -= 1
                if deleted[val] == 0:
                    del deleted[val]
            
            result = -max_heap[0] if max_heap else None
            results.append(result)
    
    return results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

import heapq
 
class MedianFinder:
    """
    Find median from a data stream using two heaps.
    
    Data structure:
    - max_heap (lower half): Contains elements <= median. Root = max of lower half.
    - min_heap (upper half): Contains elements > median. Root = min of upper half.
    
    Invariants:
    1. All elements in max_heap <= all elements in min_heap
    2. Sizes differ by at most 1: |len(max_heap) - len(min_heap)| <= 1
    
    The MAX-heap is crucial here:
    - It gives us O(1) access to the largest element of the lower half
    - Combined with min-heap's smallest of upper half, we can compute median instantly
    
    Time Complexity: O(log n) per insert, O(1) for median
    Space Complexity: O(n)
    """
    
    def __init__(self):
        # Max-heap for lower half (negate values for Python)
        self.lower_half = []  # max-heap (negated)
        # Min-heap for upper half
        self.upper_half = []  # min-heap
    
    def add_num(self, num: int) -> None:
        """Add a number to the data structure."""
        
        # Step 1: Add to appropriate heap
        if not self.lower_half or num <= -self.lower_half[0]:
            # num belongs to lower half
            heapq.heappush(self.lower_half, -num)
        else:
            # num belongs to upper half
            heapq.heappush(self.upper_half, num)
        
        # Step 2: Rebalance heaps
        # Lower half can have at most 1 more element than upper half
        if len(self.lower_half) > len(self.upper_half) + 1:
            # Move max of lower to upper
            val = -heapq.heappop(self.lower_half)
            heapq.heappush(self.upper_half, val)
        elif len(self.upper_half) > len(self.lower_half):
            # Move min of upper to lower
            val = heapq.heappop(self.upper_half)
            heapq.heappush(self.lower_half, -val)
    
    def find_median(self) -> float:
        """Return the median of all elements so far."""
        if len(self.lower_half) == len(self.upper_half):
            # Even count: average of two middle elements
            return (-self.lower_half[0] + self.upper_half[0]) / 2.0
        else:
            # Odd count: lower_half has one more element
            return float(-self.lower_half[0])
    
    def __repr__(self):
        lower = sorted(-x for x in self.lower_half)
        upper = sorted(self.upper_half)
        return f"Lower: {lower}, Upper: {upper}, Median: {self.find_median()}"
 
# Example usage
mf = MedianFinder()
for num in [2, 1, 5, 4, 3]:
    mf.add_num(num)
    print(f"Added {num}: {mf}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import heapq
from dataclasses import dataclass, field
from typing import Any, Optional
import time
 
@dataclass(order=True)
class Bid:
    """
    A bid in an auction system.
    
    Ordered by: amount (descending), then timestamp (ascending for FIFO)
    We negate amount for max-heap behavior in Python.
    """
    neg_amount: float  # Negated for max-heap: higher bid = more negative = higher priority
    timestamp: float
    bidder_id: str = field(compare=False)
    item_id: str = field(compare=False)
    metadata: Any = field(default=None, compare=False)
 
class AuctionSystem:
    """
    A max-heap-based auction system.
    
    The highest bid is always at the root, accessible in O(1).
    Bids with equal amounts are ordered by timestamp (earlier wins).
    """
    
    def __init__(self, item_id: str, reserve_price: float = 0):
        self.item_id = item_id
        self.reserve_price = reserve_price
        self._heap: list[Bid] = []
        self._bid_count = 0
    
    def place_bid(self, bidder_id: str, amount: float, 
                  metadata: Any = None) -> bool:
        """
        Place a new bid. Returns True if bid is valid (above reserve).
        """
        if amount < self.reserve_price:
            return False
        
        bid = Bid(
            neg_amount=-amount,  # Negate for max-heap
            timestamp=time.time(),
            bidder_id=bidder_id,
            item_id=self.item_id,
            metadata=metadata
        )
        heapq.heappush(self._heap, bid)
        self._bid_count += 1
        return True
    
    def get_winning_bid(self) -> Optional[Bid]:
        """Return the current highest bid (without removing)."""
        if not self._heap:
            return None
        bid = self._heap[0]
        # Return a copy with actual (non-negated) amount for clarity
        return Bid(
            neg_amount=bid.neg_amount,
            timestamp=bid.timestamp,
            bidder_id=bid.bidder_id,
            item_id=bid.item_id,
            metadata=bid.metadata
        )
    
    def get_winning_amount(self) -> float:
        """Return the current highest bid amount."""
        if not self._heap:
            return 0
        return -self._heap[0].neg_amount
    
    def get_top_k_bids(self, k: int) -> list[tuple[str, float]]:
        """Return top K bids as (bidder_id, amount) pairs."""
        # Extract and re-insert (or use nsmallest on negated heap)
        top_k = heapq.nsmallest(k, self._heap)
        return [(bid.bidder_id, -bid.neg_amount) for bid in top_k]
    
    def close_auction(self) -> Optional[tuple[str, float]]:
        """Close auction and return winner (bidder_id, amount)."""
        if not self._heap:
            return None
        winner = heapq.heappop(self._heap)
        return (winner.bidder_id, -winner.neg_amount)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

import heapq
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import defaultdict
 
@dataclass(order=True)
class LeaderboardEntry:
    """
    A leaderboard entry with score-based ordering.
    
    Negated score for max-heap behavior (highest score = most negative = top of heap).
    """
    neg_score: int
    player_id: str = field(compare=False)
    player_name: str = field(compare=False)
    
class GameLeaderboard:
    """
    Real-time leaderboard using a max-heap.
    
    Supports:
    - O(log n) score updates
    - O(1) access to current leader
    - O(k log n) for top-k retrieval
    
    Uses lazy deletion for updated scores.
    """
    
    def __init__(self, track_top_k: int = 10):
        self._heap: List[LeaderboardEntry] = []
        self._current_scores: Dict[str, int] = {}  # player_id -> current score
        self._player_names: Dict[str, str] = {}    # player_id -> name
        self._track_top_k = track_top_k
    
    def update_score(self, player_id: str, new_score: int, 
                     player_name: str = None) -> None:
        """
        Update a player's score. Creates new entry if player is new.
        
        We add a new entry without removing the old one (lazy approach).
        Old entries are filtered out when accessing the leaderboard.
        """
        self._current_scores[player_id] = new_score
        if player_name:
            self._player_names[player_id] = player_name
        
        entry = LeaderboardEntry(
            neg_score=-new_score,
            player_id=player_id,
            player_name=self._player_names.get(player_id, player_id)
        )
        heapq.heappush(self._heap, entry)
    
    def increment_score(self, player_id: str, delta: int,
                        player_name: str = None) -> int:
        """Increment player's score by delta. Returns new score."""
        current = self._current_scores.get(player_id, 0)
        new_score = current + delta
        self.update_score(player_id, new_score, player_name)
        return new_score
    
    def get_leader(self) -> Optional[tuple[str, str, int]]:
        """
        Get current leader: (player_id, player_name, score).
        Lazily removes stale entries.
        """
        while self._heap:
            entry = self._heap[0]
            current = self._current_scores.get(entry.player_id)
            
            # Check if this entry is current
            if current is not None and current == -entry.neg_score:
                return (entry.player_id, entry.player_name, current)
            
            # Stale entry: remove it
            heapq.heappop(self._heap)
        
        return None
    
    def get_top_k(self, k: int = None) -> List[tuple[str, str, int]]:
        """
        Get top K players: list of (player_id, player_name, score).
        """
        if k is None:
            k = self._track_top_k
        
        result = []
        seen = set()
        temp_removed = []
        
        while self._heap and len(result) < k:
            entry = heapq.heappop(self._heap)
            current = self._current_scores.get(entry.player_id)
            
            # Skip stale entries
            if current is None or current != -entry.neg_score:
                continue
            
            # Skip duplicates (shouldn't happen with proper updates, but safety)
            if entry.player_id in seen:
                continue
            
            seen.add(entry.player_id)
            result.append((entry.player_id, entry.player_name, current))
            temp_removed.append(entry)
        
        # Restore removed entries
        for entry in temp_removed:
            heapq.heappush(self._heap, entry)
        
        return result
    
    def get_player_rank(self, player_id: str) -> Optional[int]:
        """
        Get player's current rank (1-indexed).
        Note: O(n log n) - not efficient for frequent calls.
        """
        if player_id not in self._current_scores:
            return None
        
        player_score = self._current_scores[player_id]
        rank = 1
        for pid, score in self._current_scores.items():
            if score > player_score:
                rank += 1
        return rank