17Position and P&L Services

Thirty-three minutes into the session, the order-entry session dropped for four seconds. Four executions happened while it was down, and when it came back the strategy’s own execution reports put it 500 shares less short than the venue’s drop copy did. The position service had to decide which to believe. The answer is neither, until they agree: count every execution any source reports, once, and say that the position is pending until the sources match. Every downstream number — risk (chapter 18), limits, P&L, the books that close at night — starts from positions, and positions start from executions that arrive twice, late, out of order, and sometimes are taken back.

17.1 Positions from executions

The chapter’s service (firm.posservice) stores executions keyed by that identifier and never stores a position: every position is computed by replaying the executions in time order through Book 1’s exact keeper (firm.pnl), in which quantity and cash are integers, so that cash plus quantity times the mark, less fees, holds to the ten-thousandth of a dollar. Storing the events rather than their sum makes arrival order irrelevant, makes duplicates harmless (idempotent processing, Book 13, chapter 24) and makes every past position recomputable.

    def on_execution(self, source: str, e: Execution) -> bool:
        self.seen[source].add(e.exec_id)
        if e.exec_id in self.execs:
            return False
        self.execs[e.exec_id] = e
        return True

    def on_bust(self, exec_id: str, ts: int) -> None:
        self.busted[exec_id] = ts

    def on_correction(self, exec_id: str, ts: int, qty: int | None = None,
                      price: int | None = None) -> None:
        self.corrections.append((ts, exec_id, qty, price))

    def _effective(self, as_of=None) -> list[Execution]:
        out = {}
        for k, e in self.execs.items():
            if as_of is not None and e.ts > as_of:
                continue
            if k in self.busted and (as_of is None or self.busted[k] <= as_of):
                continue
            out[k] = e
        for ts, k, q, p in sorted(self.corrections, key=lambda c: c[0]):
            if k in out and (as_of is None or ts <= as_of):
                e = out[k]
                out[k] = replace(e, qty=e.qty if q is None else q,
                                 price=e.price if p is None else p)
        return sorted(out.values(), key=lambda e: (e.ts, e.exec_id))

    def position(self, account, symbol, as_of=None, view="trade", date=None) -> Position:
        p = Position()
        for e in self._effective(as_of):
            if e.account != account or e.symbol != symbol:
                continue
            if view == "settlement" and (date is None or e.settle_date > date):
                continue
            p.on_fill(e.side, e.qty, e.price, e.fee)
        return p

17.2 Two feeds of the same truth: sessions and drop copies

The order-entry session reports the executions of the orders it sent; the drop copy (Book 13, chapter 21) sends the firm a copy of every execution of its sessions on a separate connection, for exactly the moment the first one fails. A session that drops and logs back in asking only for new messages never receives the reports generated while it was down; a drop copy that runs a little late (50 milliseconds here) has them. The service takes both, counts each execution once, and compares the two sets: when they differ, the position it serves is the union, and its status is pending.

    def status(self, account, symbol) -> str:
        mine = {k for k, e in self.execs.items()
                if e.account == account and e.symbol == symbol}
        ids = [self.seen[s] & mine for s in self.sources]
        return "agreed" if all(i == ids[0] for i in ids) else "pending"

    def missing(self, source: str) -> set:
        others = set().union(*(self.seen[s] for s in self.sources if s != source))
        return others - self.seen[source]

17.3 Busts, corrections and late fills

A bust arrives minutes or hours after the execution; a correction changes an execution’s price or quantity. The service applies both as events with their own times, so that the position as of any moment can be computed as it was known then (the bust not yet received) or as it is known now (the trade never happened). A late fill — an execution report that arrives after the position has been read — is only a late event: the replay puts it in its place.

17.4 Intraday P&L and end-of-day marks

The chapter marks intraday at the feed’s mid and at the close at the venue’s last trade, which stands for the official close. The P&L is exact: firm.pnl keeps cash and quantity in integers, and the realised and unrealised parts under average cost (Book 1, chapter 7) add up, with fees, to the total at the mark.

17.5 Start of day and the books that close

At the end of the day the service records each position with its end-of-day mark and total; the next day’s start-of-day position is that record. The clearing firm’s statement — the executions it cleared, busts removed — is the check: the service reconciles its positions against it and lists every break before the next day starts.

17.6 Four seconds without a session

The chapter’s experiment runs a quoting strategy (500 shares a side, an inventory limit of 5 000 shares, orders left working if the session drops) for an hour on ten firm.tape sessions of Book 10’s simulator, each with a drop-copy session. At a seeded minute the order-entry session drops for four seconds; the client logs back in asking for new messages only, and operations request a replay of the missed reports a minute later. At minute 50 the venue busts the execution nearest minute 30. Three services are compared with the clearing statement: session reports only; session and drop copy merged; merged with the bust applied.

def services(d: dict) -> dict:
    """Feed the three services in arrival order; return them and the arrival log."""
    start, end = d["drop"], d["drop"] + OUTAGE
    replay = end + REPLAY_AFTER
    events = []
    for e in d["execs"]:
        lost = start <= e.ts < end
        events.append((replay if lost else e.ts, "session", e, lost))
        events.append((e.ts + DROPCOPY_LAG, "dropcopy", e, False))
    events.sort(key=lambda x: (x[0], x[1], x[2].exec_id))
    a, b, c = P.PositionService(("session",)), P.PositionService(), P.PositionService()
    for _t, src, e, replayed in events:
        if src == "session" and not replayed:
            a.on_execution(src, e)        # the session-only service never replays
        b.on_execution(src, e)
        c.on_execution(src, e)
    if d["bust"] is not None:
        c.on_bust(d["bust"].exec_id, d["open"] + BUST_NOTICE)
    return {"session": a, "merged": b, "merged+bust": c,
            "events": [(t, src, e) for t, src, e, _r in events]}

Over the ten sessions (Table 17.1), the outage costs little execution by execution — it catches executions in four sessions (one, four, three and one), because in four seconds a quoting strategy trades rarely — and the bust costs something every day. Without the drop copy the service ends the day 290 shares and $52 off on average; with it, 220 shares and $28 off, all of it the bust; with the bust applied, exact. In the sessions where the outage lost executions, the merged service reported pending for 60 to 63 seconds, until the replay; without a replay it would have reported pending until someone looked.

17.7 Tutorial: one day, three sources, one statement

17.8 Build: the position service

17.9 Exercises

17.10 Problem: Four Seconds Without a Session

17.11 Interview questions