Optimize trade PnL table updates

Real-Time Trade PnL Aggregator

In quantitative trading systems, Profit and Loss (PnL) must be tracked in real time across multiple dimensions (e.g., strategy, desk, sector, geography). A single trade can match multiple overlapping subscription criteria (views), and market prices tick rapidly. Recomputing the aggregate PnL of each view from scratch on every tick is computationally prohibitive.

Total PnL for a collection of trades is defined using standard cash-flow accounting:

Total PnL = Net Cash Flow + sum(Position(symbol) * MarkPrice(symbol))

where for each trade:

• Cash Flow = -(quantity * execution_price)
• quantity > 0 represents a buy; quantity < 0 represents a sell.
• Position(symbol) = sum(quantity) for that symbol across matching trades.

Implement the PnLAggregator class:

class PnLAggregator:
    def __init__(self):
        pass

    def register_view(self, view_id: str, tags: list[str]) -> None:
        """
        Registers a view identified by view_id.
        A trade matches this view if and only if the trade's tags contain ALL tags in the view's tags list (conjunction).
        If tags is empty, the view matches all trades.
        Views are registered before any trades or prices are added.
        """
        pass

    def add_trade(self, trade_id: str, symbol: str, quantity: int, price: float, tags: list[str]) -> None:
        """
        Records an executed trade. Multiple trades can share the same symbol.
        Updates positions and cash flows for all matching views.
        If the symbol has not received a mark price yet, initialize its mark price to this trade's price.
        """
        pass

    def update_price(self, symbol: str, price: float) -> None:
        """
        Updates the mark-to-market price for a symbol.
        """
        pass

    def get_pnl(self, view_id: str) -> float:
        """
        Returns the current total PnL for the specified view, rounded to 4 decimal places.
        If a view has no matching trades, its PnL is 0.0.
        """
        pass

Examples

agg = PnLAggregator()
agg.register_view("tech_us", ["US", "TECH"])
agg.register_view("all_us", ["US"])

# Trade 1: Buy 100 AAPL at 150.0 with tags ["US", "TECH"]
agg.add_trade("T1", "AAPL", 100, 150.0, ["US", "TECH"])
# AAPL mark price is initially 150.0
# For tech_us: Cash = -15000.0, Position = 100, PnL = -15000.0 + 100 * 150.0 = 0.0
assert agg.get_pnl("tech_us") == 0.0
assert agg.get_pnl("all_us") == 0.0

# Trade 2: Buy 50 XOM at 100.0 with tags ["US", "ENERGY"]
agg.add_trade("T2", "XOM", 50, 100.0, ["US", "ENERGY"])

# Update AAPL price to 160.0
agg.update_price("AAPL", 160.0)
# tech_us holds: 100 AAPL @ 160.0. Cash = -15000.0. PnL = 1000.0
assert agg.get_pnl("tech_us") == 1000.0
# all_us holds: 100 AAPL @ 160.0, 50 XOM @ 100.0. Cash = -20000.0. PnL = 1000.0
assert agg.get_pnl("all_us") == 1000.0

# Update XOM price to 110.0
agg.update_price("XOM", 110.0)
# tech_us is unaffected by XOM price change
assert agg.get_pnl("tech_us") == 1000.0
# all_us gains 50 * 10 = 500.0 additional PnL -> 1500.0
assert agg.get_pnl("all_us") == 1500.0

Constraints

• Number of views V <= 1,000.
• Number of trades N <= 100,000.
• Number of price updates U <= 200,000.
• Number of get_pnl calls Q 0.