Python Quant Interview Questions: 5 Worked Data Tasks

A's weighted price is (100 × 2 + 104 × 1) / 3 = 304/3, approximately 101.3333. B returns None. The ordinary mean of A's prices, 102, answers a different question.

from math import isfinite

def weighted_prices(rows):
    totals = {}
    for symbol, price, quantity in rows:
        if not isfinite(price) or price <= 0:
            raise ValueError("price must be positive and finite")
        if not isfinite(quantity) or quantity < 0:
            raise ValueError("quantity must be nonnegative and finite")
        value, volume = totals.get(symbol, (0.0, 0.0))
        totals[symbol] = (value + price * quantity,
                          volume + quantity)
    return {symbol: value / volume if volume else None
            for symbol, (value, volume) in totals.items()}

result = weighted_prices([
    ("A", 100, 2), ("A", 104, 1), ("B", 50, 0)
])
assert result == {"A": 304 / 3, "B": None}
assert weighted_prices([]) == {}

The loop takes linear time in rows and stores one pair per symbol. This reference assumes numerical input types and values small enough that intermediate products and sums remain finite. A monetary accounting system would also need an explicit precision and rounding policy.

pandas merge_asof supports backward matching, group keys and an age tolerance. Both inputs must be sorted on the time merge key. Sorting by symbol first can leave time out of order globally.

import pandas as pd

quotes = pd.DataFrame({
    "symbol": ["A", "A", "B"],
    "quote_time": [1, 4, 2],
    "price": [100, 102, 200],
})
trades = pd.DataFrame({
    "symbol": ["A", "A", "A", "B"],
    "trade_time": [0, 3, 5, 3],
})
joined = pd.merge_asof(
    trades.sort_values("trade_time"),
    quotes.sort_values("quote_time"),
    left_on="trade_time", right_on="quote_time",
    by="symbol", direction="backward", tolerance=2,
)
assert pd.isna(joined.loc[0, "price"])
# Inspect by symbol/time; merge output order follows sorted trades.
a = joined[joined.symbol == "A"].sort_values("trade_time")
assert a.price.iloc[1:].tolist() == [100, 102]
b = joined[joined.symbol == "B"]
assert b.price.tolist() == [200]
valid = joined.dropna(subset=["quote_time"])
assert (valid.quote_time <= valid.trade_time).all()
assert ((valid.trade_time - valid.quote_time) <= 2).all()

Retaining the quote timestamp makes the information boundary auditable. Test a trade before all quotes, one with an overly old quote, a different symbol and an exact timestamp match. If the original trade order matters, add a row identifier before sorting and restore that order after the join.

import pandas as pd

prices = pd.Series([100, 102, 101, 105])
feature = prices.shift(1).rolling(2, min_periods=2).mean()
assert feature.iloc[:2].isna().all()
assert feature.iloc[2:].tolist() == [101.0, 101.5]

The shift expresses the information boundary. A plain rolling mean includes the current row. Whether that is leakage depends on when the prediction is made; here the contract explicitly excludes that observation.

With multiple symbols, sort each symbol's observations chronologically and shift within the group. A global shift can import the last price of one symbol into the next symbol's feature. A simple grouped implementation is:

def add_feature(frame):
    ordered = frame.sort_values(["symbol", "time"]).copy()
    ordered["feature"] = ordered.groupby("symbol")["price"].transform(
        lambda s: s.shift(1).rolling(2, min_periods=2).mean()
    )
    return ordered

from collections import deque
from math import isfinite

def rolling_means(values, k):
    if type(k) is not int or k <= 0:
        raise ValueError("k must be a positive integer")
    window = deque()
    total = 0.0
    for value in values:
        if not isfinite(value):
            raise ValueError("values must be finite")
        if len(window) == k:
            total -= window.popleft()
        window.append(value)
        total += value
        yield total / len(window)

assert list(rolling_means([1, 3, 8, 4], 2)) == [1, 2, 5.5, 6]
assert list(rolling_means([], 2)) == []
assert list(rolling_means([7, 9], 1)) == [7, 9]

Each update performs constant queue work and uses at most k stored values. This reference assumes intermediate sums remain finite; finite inputs can still overflow a floating-point running total. Compare the output against a simple implementation that slices and recomputes each window. Floating-point addition and subtraction accumulate rounding error; discuss periodic recomputation or a more careful numerical method if accuracy requirements demand it.

import numpy as np

returns = np.array([[0.01, 0.02], [-0.01, 0.03], [0, -0.02]])
weights = np.array([0.25, 0.75])
portfolio = (returns * weights).sum(axis=1)
np.testing.assert_allclose(portfolio, [0.0175, 0.02, -0.015])
assert portfolio.shape == (3,)

The weights broadcast across observations, then summing across columns produces three results. With weights shaped (2, 1), the multiplication fails: the leading dimensions 3 and 2 are incompatible. NumPy's broadcasting documentation explains the right-to-left dimension comparison and warns that some broadcasted calculations create inefficient intermediate arrays.