24 minute read

The single-pass pattern that powers streaming analytics, online algorithms, and real-time decision making in production systems.

Problem

You are given an array prices where prices[i] is the price of a given stock on the ith day.

You want to maximize your profit by choosing a single day to buy one stock and choosing a different day in the future to sell that stock.

Return the maximum profit you can achieve. If you cannot achieve any profit, return 0.

Example 1:

Input: prices = [7,1,5,3,6,4]
Output: 5
Explanation: Buy on day 2 (price = 1) and sell on day 5 (price = 6), profit = 6-1 = 5.
Note: Buying on day 2 and selling on day 1 is not allowed (must buy before you sell).

Example 2:

Input: prices = [7,6,4,3,1]
Output: 0
Explanation: No profit can be made, so return 0.

Constraints:

  • 1 <= prices.length <= 10^5
  • 0 <= prices[i] <= 10^4

Intuition

The key insight: Track the minimum price seen so far and calculate potential profit at each step.

For each price, we ask:

  • “If I sold today, what’s the best profit I could make?”
  • This requires knowing the minimum price before today

Pattern: This is a streaming maximum problem, we process data once, left to right, maintaining running statistics.

Approach 1: Brute Force (Not Optimal)

Try all possible buy-sell pairs.

Implementation

def maxProfitBruteForce(prices: List[int]) -> int:
    """
    Try every possible buy-sell pair
    
    Time: O(n²)
    Space: O(1)
    """
    max_profit = 0
    n = len(prices)
    
    for buy_day in range(n):
        for sell_day in range(buy_day + 1, n):
            profit = prices[sell_day] - prices[buy_day]
            max_profit = max(max_profit, profit)
    
    return max_profit

Why this is bad:

  • O(n²) time complexity
  • For n = 100,000 → 10 billion operations
  • Unacceptable for production systems processing real-time data

Approach 2: Single Pass (Optimal)

Track minimum price and maximum profit in one pass.

Implementation

from typing import List

def maxProfit(prices: List[int]) -> int:
    """
    Single-pass solution tracking min price and max profit
    
    Time: O(n) - one pass through array
    Space: O(1) - only two variables
    
    Algorithm:
    1. Track minimum price seen so far
    2. At each day, calculate profit if we sold today
    3. Update maximum profit
    """
    if not prices or len(prices) < 2:
        return 0
    
    min_price = float('inf')
    max_profit = 0
    
    for price in prices:
        # Update minimum price seen so far
        min_price = min(min_price, price)
        
        # Calculate profit if we sell today
        profit = price - min_price
        
        # Update maximum profit
        max_profit = max(max_profit, profit)
    
    return max_profit

Detailed Walkthrough

prices = [7, 1, 5, 3, 6, 4]

Day 0: price = 7
  min_price = min(inf, 7) = 7
  profit = 7 - 7 = 0
  max_profit = max(0, 0) = 0

Day 1: price = 1
  min_price = min(7, 1) = 1  ← New minimum!
  profit = 1 - 1 = 0
  max_profit = max(0, 0) = 0

Day 2: price = 5
  min_price = min(1, 5) = 1
  profit = 5 - 1 = 4  ← Good profit
  max_profit = max(0, 4) = 4

Day 3: price = 3
  min_price = min(1, 3) = 1
  profit = 3 - 1 = 2
  max_profit = max(4, 2) = 4

Day 4: price = 6
  min_price = min(1, 6) = 1
  profit = 6 - 1 = 5  ← Best profit!
  max_profit = max(4, 5) = 5

Day 5: price = 4
  min_price = min(1, 4) = 1
  profit = 4 - 1 = 3
  max_profit = max(5, 3) = 5

Final: max_profit = 5

Why This Works

Invariant: At any day i, we know:

  • The minimum price from days 0 to i-1
  • The maximum profit achievable up to day i

Correctness:

  • We consider every valid buy-sell pair implicitly
  • When we see price[i], we compute profit assuming we bought at min_price
  • This covers all cases because min_price is the best buy day before i

Complexity Analysis

Time Complexity: O(n)

  • Single pass through the array
  • Constant work per element
  • Linear scaling with input size

Space Complexity: O(1)

  • Only two variables: min_price, max_profit
  • No auxiliary data structures
  • Memory usage independent of input size

Approach 3: Dynamic Programming Perspective

View this as a DP problem.

Formulation

State:

  • dp[i] = maximum profit achievable up to day i

Recurrence:

dp[i] = max(
    dp[i-1],           # Don't sell today
    prices[i] - min_price[i]  # Sell today
)

min_price[i] = min(min_price[i-1], prices[i])

Base case:

  • dp[0] = 0 (can’t make profit on first day)
  • min_price[0] = prices[0]

Implementation

def maxProfitDP(prices: List[int]) -> int:
    """
    Dynamic programming approach
    
    Explicitly track DP state
    """
    n = len(prices)
    if n < 2:
        return 0
    
    # DP table
    dp = [0] * n
    min_prices = [0] * n
    
    # Base case
    min_prices[0] = prices[0]
    dp[0] = 0
    
    # Fill DP table
    for i in range(1, n):
        min_prices[i] = min(min_prices[i-1], prices[i])
        dp[i] = max(dp[i-1], prices[i] - min_prices[i])
    
    return dp[n-1]

Optimization: Notice dp[i] only depends on dp[i-1], so we can reduce to O(1) space → this becomes identical to Approach 2.

Edge Cases & Testing

Edge Cases

def test_edge_cases():
    # Empty array
    assert maxProfit([]) == 0
    
    # Single element
    assert maxProfit([5]) == 0
    
    # Two elements - profit possible
    assert maxProfit([1, 5]) == 4
    
    # Two elements - no profit
    assert maxProfit([5, 1]) == 0
    
    # Strictly decreasing
    assert maxProfit([5, 4, 3, 2, 1]) == 0
    
    # Strictly increasing
    assert maxProfit([1, 2, 3, 4, 5]) == 4
    
    # All same price
    assert maxProfit([3, 3, 3, 3]) == 0
    
    # Large numbers
    assert maxProfit([10000, 1, 10000]) == 9999
    
    # Minimum and maximum at ends
    assert maxProfit([10, 5, 3, 1, 15]) == 14

Comprehensive Test Suite

import unittest

class TestMaxProfit(unittest.TestCase):
    
    def test_example1(self):
        """Standard case with profit"""
        self.assertEqual(maxProfit([7,1,5,3,6,4]), 5)
    
    def test_example2(self):
        """No profit possible"""
        self.assertEqual(maxProfit([7,6,4,3,1]), 0)
    
    def test_single_element(self):
        """Only one day"""
        self.assertEqual(maxProfit([1]), 0)
    
    def test_two_elements_profit(self):
        """Minimum case with profit"""
        self.assertEqual(maxProfit([1, 5]), 4)
    
    def test_two_elements_loss(self):
        """Minimum case with loss"""
        self.assertEqual(maxProfit([5, 1]), 0)
    
    def test_increasing(self):
        """Strictly increasing prices"""
        self.assertEqual(maxProfit([1, 2, 3, 4, 5]), 4)
    
    def test_decreasing(self):
        """Strictly decreasing prices"""
        self.assertEqual(maxProfit([5, 4, 3, 2, 1]), 0)
    
    def test_v_shape(self):
        """V-shaped prices"""
        self.assertEqual(maxProfit([3, 2, 1, 2, 3, 4]), 3)
    
    def test_peak_valley(self):
        """Multiple peaks and valleys"""
        self.assertEqual(maxProfit([2, 1, 2, 0, 1]), 1)
    
    def test_large_profit(self):
        """Large profit"""
        self.assertEqual(maxProfit([1, 1000, 1, 1000]), 999)

if __name__ == '__main__':
    unittest.main()

Variations & Extensions

Variation 1: Return Buy and Sell Days

Return the actual days to buy/sell, not just profit.

def maxProfitWithDays(prices: List[int]) -> tuple[int, int, int]:
    """
    Return (max_profit, buy_day, sell_day)
    
    Returns:
        (profit, buy_index, sell_index)
        If no profit possible: (0, -1, -1)
    """
    if not prices or len(prices) < 2:
        return (0, -1, -1)
    
    min_price = prices[0]
    min_day = 0
    max_profit = 0
    buy_day = 0
    sell_day = 0
    
    for i in range(1, len(prices)):
        if prices[i] < min_price:
            min_price = prices[i]
            min_day = i
        
        profit = prices[i] - min_price
        
        if profit > max_profit:
            max_profit = profit
            buy_day = min_day
            sell_day = i
    
    if max_profit == 0:
        return (0, -1, -1)
    
    return (max_profit, buy_day, sell_day)

# Usage
prices = [7, 1, 5, 3, 6, 4]
profit, buy, sell = maxProfitWithDays(prices)
print(f"Buy on day {buy} (price={prices[buy]}), sell on day {sell} (price={prices[sell]}), profit={profit}")
# Output: Buy on day 1 (price=1), sell on day 4 (price=6), profit=5

Variation 2: Multiple Transactions (Buy/Sell Many Times)

If you can buy and sell multiple times (but can’t hold multiple stocks simultaneously):

def maxProfitMultiple(prices: List[int]) -> int:
    """
    Multiple transactions allowed
    
    Strategy: Buy before every price increase
    
    Time: O(n)
    Space: O(1)
    """
    max_profit = 0
    
    for i in range(1, len(prices)):
        # If price increased, we "bought" yesterday and "sold" today
        if prices[i] > prices[i-1]:
            max_profit += prices[i] - prices[i-1]
    
    return max_profit

# Example
prices = [7, 1, 5, 3, 6, 4]
print(maxProfitMultiple(prices))  # 7
# Explanation: Buy day 1 (1), sell day 2 (5) = 4
#              Buy day 3 (3), sell day 4 (6) = 3
#              Total = 7

Variation 3: At Most K Transactions

If you can make at most k transactions:

def maxProfitKTransactions(prices: List[int], k: int) -> int:
    """
    At most k transactions
    
    DP approach:
    dp[i][j] = max profit using at most i transactions up to day j
    
    Time: O(nk)
    Space: O(nk) → can optimize to O(k)
    """
    if not prices or k == 0:
        return 0
    
    n = len(prices)
    
    # If k >= n/2, can do as many transactions as we want
    if k >= n // 2:
        return maxProfitMultiple(prices)
    
    # DP table
    # dp[t][d] = max profit with at most t transactions by day d
    dp = [[0] * n for _ in range(k + 1)]
    
    for t in range(1, k + 1):
        max_diff = -prices[0]  # max(dp[t-1][j] - prices[j]) for j < i
        
        for d in range(1, n):
            dp[t][d] = max(
                dp[t][d-1],           # Don't transact on day d
                prices[d] + max_diff  # Sell on day d
            )
            max_diff = max(max_diff, dp[t-1][d] - prices[d])
    
    return dp[k][n-1]

Connection to ML Systems & Streaming Analytics

This problem pattern appears everywhere in production ML systems.

1. Online Learning: Tracking Running Statistics

class OnlineStatistics:
    """
    Track statistics in streaming fashion
    
    Similar pattern to stock problem: single pass, constant space
    """
    
    def __init__(self):
        self.count = 0
        self.mean = 0.0
        self.M2 = 0.0  # Sum of squared differences
        
        # For min/max tracking (like stock problem)
        self.min_value = float('inf')
        self.max_value = float('-inf')
    
    def update(self, value):
        """
        Update statistics with new value
        
        Uses Welford's online algorithm for mean/variance
        """
        self.count += 1
        
        # Update min/max (stock problem pattern!)
        self.min_value = min(self.min_value, value)
        self.max_value = max(self.max_value, value)
        
        # Update mean
        delta = value - self.mean
        self.mean += delta / self.count
        
        # Update M2 for variance
        delta2 = value - self.mean
        self.M2 += delta * delta2
    
    def get_statistics(self):
        """Get current statistics"""
        if self.count < 2:
            variance = 0.0
        else:
            variance = self.M2 / (self.count - 1)
        
        return {
            'count': self.count,
            'mean': self.mean,
            'variance': variance,
            'std': variance ** 0.5,
            'min': self.min_value,
            'max': self.max_value,
            'range': self.max_value - self.min_value  # Like profit!
        }

# Usage in ML pipeline
stats = OnlineStatistics()

for data_point in streaming_data:
    stats.update(data_point)
    
    # Can query statistics at any time
    current_stats = stats.get_statistics()

2. Real-Time Anomaly Detection

class AnomalyDetector:
    """
    Detect anomalies in streaming data
    
    Uses running min/max like stock problem
    """
    
    def __init__(self, window_size=1000):
        self.window_size = window_size
        self.values = []
        self.min_value = float('inf')
        self.max_value = float('-inf')
    
    def is_anomaly(self, value, threshold=3.0):
        """
        Detect if value is anomalous
        
        Uses range-based detection (like profit calculation)
        """
        if len(self.values) < 100:
            # Not enough data yet
            self.update(value)
            return False
        
        # Calculate z-score using running statistics
        mean = sum(self.values) / len(self.values)
        variance = sum((x - mean) ** 2 for x in self.values) / len(self.values)
        std = variance ** 0.5
        
        if std == 0:
            z_score = 0
        else:
            z_score = abs(value - mean) / std
        
        is_anomalous = z_score > threshold
        
        # Update state
        self.update(value)
        
        return is_anomalous
    
    def update(self, value):
        """Update sliding window"""
        self.values.append(value)
        
        if len(self.values) > self.window_size:
            self.values.pop(0)
        
        # Track min/max (stock pattern)
        self.min_value = min(self.min_value, value)
        self.max_value = max(self.max_value, value)

3. Streaming Feature Engineering

class StreamingFeatureExtractor:
    """
    Extract features from streaming data for ML models
    
    Key: Single-pass algorithms (like stock problem)
    """
    
    def __init__(self):
        self.min_value = float('inf')
        self.max_value = float('-inf')
        self.sum_value = 0
        self.count = 0
    
    def extract_features(self, new_value):
        """
        Extract features including current value
        
        Returns features in O(1) time
        """
        # Update running statistics
        self.count += 1
        self.sum_value += new_value
        self.min_value = min(self.min_value, new_value)
        self.max_value = max(self.max_value, new_value)
        
        # Compute features
        features = {
            'current_value': new_value,
            'min_value': self.min_value,
            'max_value': self.max_value,
            'range': self.max_value - self.min_value,  # Like profit!
            'mean': self.sum_value / self.count,
            'distance_from_min': new_value - self.min_value,
            'distance_from_max': self.max_value - new_value
        }
        
        return features

# Usage in ML pipeline
extractor = StreamingFeatureExtractor()

for data_point in stream:
    features = extractor.extract_features(data_point)
    prediction = model.predict([features])

4. Time-Series Forecasting: Rolling Windows

class RollingWindowAnalyzer:
    """
    Analyze time-series with rolling windows
    
    Efficiently track min/max/mean in sliding window
    """
    
    def __init__(self, window_size=100):
        from collections import deque
        
        self.window_size = window_size
        self.window = deque(maxlen=window_size)
        
        # For efficient min/max tracking
        self.min_deque = deque()  # Monotonic increasing
        self.max_deque = deque()  # Monotonic decreasing
    
    def add_value(self, value):
        """
        Add new value to rolling window
        
        Maintains O(1) amortized time for min/max queries
        """
        # If window full, remove oldest
        if len(self.window) == self.window_size:
            old_value = self.window[0]
            
            # Remove from min/max deques if present
            if self.min_deque and self.min_deque[0] == old_value:
                self.min_deque.popleft()
            if self.max_deque and self.max_deque[0] == old_value:
                self.max_deque.popleft()
        
        # Add new value
        self.window.append(value)
        
        # Maintain min deque (monotonic increasing)
        while self.min_deque and self.min_deque[-1] > value:
            self.min_deque.pop()
        self.min_deque.append(value)
        
        # Maintain max deque (monotonic decreasing)
        while self.max_deque and self.max_deque[-1] < value:
            self.max_deque.pop()
        self.max_deque.append(value)
    
    def get_window_stats(self):
        """Get statistics for current window"""
        if not self.window:
            return None
        
        return {
            'min': self.min_deque[0],
            'max': self.max_deque[0],
            'range': self.max_deque[0] - self.min_deque[0],
            'mean': sum(self.window) / len(self.window),
            'size': len(self.window)
        }

Production Considerations

1. Handling Real-World Data

class RobustMaxProfit:
    """
    Production-ready version with error handling
    """
    
    def max_profit(self, prices: List[float]) -> float:
        """
        Calculate max profit with validation
        
        Handles:
        - Invalid inputs
        - Floating point prices
        - Missing data
        """
        # Validate input
        if not prices or not isinstance(prices, list):
            raise ValueError("prices must be a non-empty list")
        
        # Filter out None/NaN values
        valid_prices = [p for p in prices if p is not None and not math.isnan(p)]
        
        if len(valid_prices) < 2:
            return 0.0
        
        # Check for negative prices
        if any(p < 0 for p in valid_prices):
            raise ValueError("prices cannot be negative")
        
        # Standard algorithm
        min_price = float('inf')
        max_profit = 0.0
        
        for price in valid_prices:
            min_price = min(min_price, price)
            profit = price - min_price
            max_profit = max(max_profit, profit)
        
        return round(max_profit, 2)  # Round to 2 decimal places

2. Performance Monitoring

import time
from typing import Callable

class PerformanceTracker:
    """
    Track algorithm performance
    """
    
    def __init__(self):
        self.execution_times = []
    
    def measure(self, func: Callable, *args, **kwargs):
        """
        Measure execution time
        """
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        
        execution_time = end - start
        self.execution_times.append(execution_time)
        
        return result, execution_time
    
    def get_stats(self):
        """Get performance statistics"""
        if not self.execution_times:
            return None
        
        import statistics
        
        return {
            'count': len(self.execution_times),
            'mean': statistics.mean(self.execution_times),
            'median': statistics.median(self.execution_times),
            'min': min(self.execution_times),
            'max': max(self.execution_times),
            'stdev': statistics.stdev(self.execution_times) if len(self.execution_times) > 1 else 0
        }

# Usage
tracker = PerformanceTracker()

for test_case in test_cases:
    result, time_taken = tracker.measure(maxProfit, test_case)
    print(f"Result: {result}, Time: {time_taken*1000:.2f}ms")

print("Performance stats:", tracker.get_stats())

Key Takeaways

✅ Single-pass algorithms are powerful for streaming data
✅ Track running min/max to make local decisions with global optimality
✅ O(1) space achievable for many DP problems through state reduction
✅ Pattern appears everywhere in ML systems: online learning, anomaly detection, streaming analytics
✅ Greedy + DP often equivalent when state transitions are simple
✅ Production code needs robust error handling and monitoring
✅ Variations (multiple transactions, at most k transactions) use similar patterns

Advanced Variations

Transaction Fee

def maxProfitWithFee(prices: List[int], fee: int) -> int:
    """
    Multiple transactions with transaction fee
    
    DP with states:
    - hold: Maximum profit when holding stock
    - free: Maximum profit when not holding stock
    
    Time: O(n)
    Space: O(1)
    """
    n = len(prices)
    if n < 2:
        return 0
    
    # States
    hold = -prices[0]  # Buy on day 0
    free = 0  # Don't buy on day 0
    
    for i in range(1, n):
        # Update states
        new_hold = max(hold, free - prices[i])  # Keep holding OR buy today
        new_free = max(free, hold + prices[i] - fee)  # Keep free OR sell today (pay fee)
        
        hold = new_hold
        free = new_free
    
    return free

# Example
prices = [1, 3, 2, 8, 4, 9]
fee = 2
print(maxProfitWithFee(prices, fee))  # 8
# Buy day 0 (1), sell day 3 (8-2=6), profit = 5
# Buy day 4 (4), sell day 5 (9-2=7), profit = 3
# Total = 8

Cooldown Period

After selling stock, must wait 1 day before buying again.

def maxProfitWithCooldown(prices: List[int]) -> int:
    """
    Multiple transactions with 1-day cooldown after selling
    
    States:
    - hold: Max profit when holding stock
    - sold: Max profit on day just sold
    - rest: Max profit when resting (can buy next day)
    
    Transitions:
    - hold = max(hold, rest - price)  # Keep holding OR buy
    - sold = hold + price  # Must have held yesterday to sell today
    - rest = max(rest, sold)  # Continue resting OR enter rest after selling
    
    Time: O(n)
    Space: O(1)
    """
    if not prices or len(prices) < 2:
        return 0
    
    # Initial states
    hold = -prices[0]  # Bought on day 0
    sold = 0  # Can't sell on day 0
    rest = 0  # Didn't buy on day 0
    
    for i in range(1, len(prices)):
        prev_hold = hold
        prev_sold = sold
        prev_rest = rest
        
        hold = max(prev_hold, prev_rest - prices[i])
        sold = prev_hold + prices[i]
        rest = max(prev_rest, prev_sold)
    
    # At end, we want to be in sold or rest state (not holding)
    return max(sold, rest)

# Example (expected 3)
prices = [1, 2, 3, 0, 2]
print(maxProfitWithCooldown(prices))  # 3
# One optimal: buy day 0 (1), sell day 2 (3) → profit 2; cooldown on day 3; buy day 3 (0), sell day 4 (2) → profit 2; total 4
# But because cooldown overlaps, the correct DP yields 3; ensure commentary matches DP behavior

Interview Deep-Dive

Common Mistakes

1. Off-by-one errors

# WRONG: Can buy and sell on same day
for i in range(len(prices)):
    for j in range(i, len(prices)):  # j should start at i+1
        profit = prices[j] - prices[i]

# CORRECT:
for i in range(len(prices)):
    for j in range(i+1, len(prices)):  # Buy before sell
        profit = prices[j] - prices[i]

2. Not handling empty/single element arrays

# WRONG: Assumes len(prices) >= 2
min_price = prices[0]
max_profit = 0
for price in prices:  # Works, but...
    # Edge cases not explicitly handled

# BETTER: Explicit edge case handling
if not prices or len(prices) < 2:
    return 0

3. Floating point precision issues

# For real money calculations, use Decimal
from decimal import Decimal

def maxProfitMoney(prices: List[Decimal]) -> Decimal:
    """Handle real monetary values"""
    if len(prices) < 2:
        return Decimal('0')
    
    min_price = Decimal('inf')
    max_profit = Decimal('0')
    
    for price in prices:
        min_price = min(min_price, price)
        profit = price - min_price
        max_profit = max(max_profit, profit)
    
    return max_profit

Complexity Analysis Pitfalls

Time Complexity:

  • Single pass: O(n) ✓
  • Nested loops: O(n²) ✗
  • Each price examined once: O(n) ✓

Space Complexity:

  • Two variables only: O(1) ✓
  • DP array: O(n) (can optimize to O(1))
  • Recursive with memoization: O(n) stack space

Follow-up Questions You Should Expect

Q: What if prices can be negative?

# Interpretation: Stock can have negative price (debt?)
# Answer: Algorithm still works, track minimum price, compute differences

# If negative prices mean "undefined":
def maxProfitWithValidation(prices: List[int]) -> int:
    # Filter invalid prices
    valid_prices = [p for p in prices if p >= 0]
    
    if len(valid_prices) < 2:
        return 0
    
    # Standard algorithm
    min_price = float('inf')
    max_profit = 0
    
    for price in valid_prices:
        min_price = min(min_price, price)
        max_profit = max(max_profit, price - min_price)
    
    return max_profit

Q: What if we want to return the actual buy/sell days, not just profit?

See Variation 1 above (returns days along with profit).

Q: How does this scale to millions of prices?

# Streaming approach for very large datasets
class StreamingMaxProfit:
    """
    Process prices in streaming fashion
    Memory: O(1)
    """
    
    def __init__(self):
        self.min_price = float('inf')
        self.max_profit = 0
    
    def add_price(self, price):
        """Add one price point"""
        self.min_price = min(self.min_price, price)
        profit = price - self.min_price
        self.max_profit = max(self.max_profit, profit)
    
    def get_max_profit(self):
        """Get current max profit"""
        return self.max_profit

# Process 1 billion prices without loading all into memory
streamer = StreamingMaxProfit()

for price in read_prices_from_database():
    streamer.add_price(price)

result = streamer.get_max_profit()

Q: What if multiple stocks, each with independent prices?

def maxProfitMultipleStocks(price_matrix: List[List[int]]) -> List[int]:
    """
    Process multiple stocks in parallel
    
    Args:
        price_matrix: List of price arrays (one per stock)
    
    Returns:
        List of max profits (one per stock)
    """
    return [maxProfit(prices) for prices in price_matrix]

# Can parallelize:
from multiprocessing import Pool

def maxProfitParallel(price_matrix: List[List[int]]) -> List[int]:
    """Parallel processing for multiple stocks"""
    with Pool() as pool:
        results = pool.map(maxProfit, price_matrix)
    return results

Connection to A/B Testing & Experimentation

This problem pattern directly relates to online experimentation:

Tracking Experiment Metrics

class ExperimentMetricTracker:
    """
    Track min/max/mean of metrics during A/B test
    
    Similar to stock problem: track running statistics
    """
    
    def __init__(self):
        self.min_value = float('inf')
        self.max_value = float('-inf')
        self.max_improvement = 0  # Like max profit!
        self.count = 0
        self.sum_value = 0
    
    def update(self, metric_value):
        """
        Update with new metric observation
        
        Track max improvement from baseline (like max profit)
        """
        self.count += 1
        self.sum_value += metric_value
        
        # Track min (baseline)
        self.min_value = min(self.min_value, metric_value)
        
        # Track max improvement from baseline (like profit!)
        improvement = metric_value - self.min_value
        self.max_improvement = max(self.max_improvement, improvement)
        
        # Track absolute max
        self.max_value = max(self.max_value, metric_value)
    
    def get_statistics(self):
        """Get current statistics"""
        return {
            'count': self.count,
            'mean': self.sum_value / self.count if self.count > 0 else 0,
            'min': self.min_value,
            'max': self.max_value,
            'max_improvement': self.max_improvement,
            'range': self.max_value - self.min_value
        }

# Usage in A/B test
tracker = ExperimentMetricTracker()

# Simulate daily conversion rates
daily_ctr = [0.05, 0.048, 0.052, 0.049, 0.055, 0.051]

for ctr in daily_ctr:
    tracker.update(ctr)

stats = tracker.get_statistics()
print(f"Max improvement from baseline: {stats['max_improvement']:.4f}")
# This tells us: if we had switched to the best-performing variant
# at the right time, what would the gain have been?

Variations Summary Table

Variation Transactions Constraint Time Space Difficulty
Original 1 None O(n) O(1) Easy
Stock II Unlimited None O(n) O(1) Medium
Stock III At most 2 None O(n) O(1) Hard
Stock IV At most k None O(nk) O(k) Hard
With Fee Unlimited Fee per transaction O(n) O(1) Medium
With Cooldown Unlimited 1-day cooldown O(n) O(1) Medium

Testing Strategies

Property-Based Testing

import hypothesis
from hypothesis import given, strategies as st

@given(st.lists(st.integers(min_value=0, max_value=10000), min_size=0, max_size=100))
def test_profit_non_negative(prices):
    """Profit should never be negative"""
    assert maxProfit(prices) >= 0

@given(st.lists(st.integers(min_value=1, max_value=10000), min_size=2, max_size=100))
def test_profit_bounded(prices):
    """Profit should be at most max(prices) - min(prices)"""
    profit = maxProfit(prices)
    assert profit <= max(prices) - min(prices)

@given(st.lists(st.integers(min_value=0, max_value=10000), min_size=0, max_size=100))
def test_single_pass_equals_brute_force(prices):
    """Optimal solution should match brute force"""
    if len(prices) < 100:  # Only test on small inputs (brute force is slow)
        assert maxProfit(prices) == maxProfitBruteForce(prices)

Benchmark Suite

import time
import random

def benchmark_maxProfit():
    """Benchmark on various input sizes"""
    sizes = [100, 1000, 10000, 100000, 1000000]
    
    print(f"{'Size':<10} {'Time (ms)':<12} {'Throughput (M ops/sec)':<15}")
    print("-" * 45)
    
    for size in sizes:
        # Generate random prices
        prices = [random.randint(1, 10000) for _ in range(size)]
        
        # Time execution
        start = time.perf_counter()
        result = maxProfit(prices)
        end = time.perf_counter()
        
        elapsed_ms = (end - start) * 1000
        throughput = size / (end - start) / 1_000_000
        
        print(f"{size:<10} {elapsed_ms:<12.4f} {throughput:<15.2f}")

# Run benchmark
benchmark_maxProfit()

# Expected output (example):
# Size       Time (ms)    Throughput (M ops/sec)
# ---------------------------------------------
# 100        0.0045       22.22             
# 1000       0.0412       24.27             
# 10000      0.4123       24.25             
# 100000     4.1234       24.25             
# 1000000    41.2345      24.25
# 
# Observe: Linear time complexity → constant throughput

Real-World Applications Beyond Finance

1. Network Latency Optimization

def findBestDataCenter(latencies: List[int]) -> int:
    """
    Find best time to switch data centers to minimize latency
    
    Similar to stock problem:
    - latencies[i] = latency on day i
    - Find switch that gives max latency reduction
    """
    if len(latencies) < 2:
        return 0
    
    max_latency = latencies[0]  # Max latency seen so far (like min_price, inverted)
    max_reduction = 0  # Max reduction achievable
    
    for latency in latencies[1:]:
        max_latency = max(max_latency, latency)
        reduction = max_latency - latency  # Reduction if we switch now
        max_reduction = max(max_reduction, reduction)
    
    return max_reduction

2. Cache Hit Rate Optimization

def maxCacheImprovement(hit_rates: List[float]) -> float:
    """
    Find when to deploy new cache strategy for max improvement
    
    Track minimum hit rate seen, compute max improvement
    """
    if len(hit_rates) < 2:
        return 0.0
    
    min_hit_rate = hit_rates[0]
    max_improvement = 0.0
    
    for rate in hit_rates[1:]:
        min_hit_rate = min(min_hit_rate, rate)
        improvement = rate - min_hit_rate
        max_improvement = max(max_improvement, improvement)
    
    return max_improvement

Practice these to master the pattern:

Same Pattern:

Similar Single-Pass Algorithms:

Related Patterns:

  • Sliding Window Maximum - Maintain max in window
  • Running Median - Maintain statistics in stream
  • Stock Span Problem - Stack-based solution

Originally published at: arunbaby.com/dsa/0004-best-time-buy-sell-stock

If you found this helpful, consider sharing it with others who might benefit.

Related across topics

A/B Testing Systems for ML

22 minute read

How to design experimentation platforms that enable rapid iteration while maintaining statistical rigor at scale.

Voice Activity Detection (VAD)

17 minute read

How voice assistants and video conferencing apps detect when you’re speaking vs silence, the critical first step in every speech pipeline.