按策略维度的盯市 P&L 簿记
Per-Strategy P&L Bookkeeping with Mark-to-Market
开始编码多策略执行平台把 N 个独立算法策略(做市、统计套利、动量、组合对冲……)的成交流通过一个共享的下单路由汇聚出去,然后必须把 P&L 按策略归因回去——以支撑业绩报告与风控限额。簿记模块夹在成交流和看板之间:它消费一个按时间序排列的事件日志——{type: "trade", strategy_id, qty, price} 的成交事件,与 {type: "mark", mark_price} 的盯市 tick(共享给所有持有该标的的策略,单一 mark 价格广播)交错——并在最后输出每个策略的 running position、加权平均成本、累计已实现 P&L 与最近一次的未实现 P&L 快照。
请实现 solution(events) -> dict[str, dict]。按顺序遍历事件,按以下规则维护每策略的状态:
- **
{type: "trade", strategy_id, qty, price}**——qty带符号(正=买,负=卖)。先在 dict 中查找(或创建)该策略行;若qty == 0,no-op。否则按下面的规则更新 position 与 avg_cost:(a) 从 0 开仓 →avg_cost = price;(b) 同向加仓 → 按绝对量加权平均;(c) 反向减仓但未跨过 0 → 计入realized_pnl += closed_qty * (price - avg_cost) * sign(position),avg_cost不变(成本基础只在开仓/加仓时变);(d) 跨过 0(close-and-reverse)→ 先把整段已有仓位平掉计入 realized,再把avg_cost重置为price作为新方向残余仓位的成本。当 position 恰好等于 0 时,avg_cost也归零。 - **
{type: "mark", mark_price}**——对当下 state dict 里所有的策略,置last_unrealized_pnl = position * (mark_price - avg_cost)(position 为 0 时即为 0)。尚未被 trade 创建的策略不会被追溯创建。 - **
{type: "ignore", ...}** 或其它未知 type——静默跳过(用于「未知事件类型」的健壮性测试)。
输出:一个 dict,strategy_id -> {position: float, avg_cost: float, realized_pnl: float, last_unrealized_pnl: float}。position 为 0 时 avg_cost = 0.0;尚无任何 mark 事件作用过该策略时 last_unrealized_pnl = 0.0。
示例
solution([{"type": "trade", "strategy_id": "S1", "qty": 10, "price": 100.0}, {"type": "trade", "strategy_id": "S1", "qty": -15, "price": 110.0}]) 返回 {"S1": {"position": -5.0, "avg_cost": 110.0, "realized_pnl": 100.0, "last_unrealized_pnl": 0.0}}。先多 10@100 开仓;再卖 15@110 跨过 0——平掉 10 多头,realized = 10 * (110 - 100) * +1 = 100;残余 -5 反向开仓,avg_cost = 110。
solution([{"type": "trade", "strategy_id": "S1", "qty": 10, "price": 100.0}, {"type": "trade", "strategy_id": "S2", "qty": -5, "price": 200.0}, {"type": "mark", "mark_price": 150.0}]) 返回 {"S1": {"position": 10.0, "avg_cost": 100.0, "realized_pnl": 0.0, "last_unrealized_pnl": 500.0}, "S2": {"position": -5.0, "avg_cost": 200.0, "realized_pnl": 0.0, "last_unrealized_pnl": 250.0}}。两个反向策略共享一次 mark:S1(多 10@100)盯市于 150 → unrealized = 10*(150-100)=500;S2(空 -5@200)盯市于 150 → unrealized = -5*(150-200)=250(空头在价跌中盈利)。
solution([{"type": "mark", "mark_price": 100.0}, {"type": "trade", "strategy_id": "S1", "qty": 10, "price": 110.0}]) 返回 {"S1": {"position": 10.0, "avg_cost": 110.0, "realized_pnl": 0.0, "last_unrealized_pnl": 0.0}}。mark 发生在任何 trade 之前——彼时 dict 里还没有任何策略可被盯市;S1 在随后的 trade 才被创建,并因「之后没有再发生过 mark」而保持 last_unrealized_pnl = 0.0。
函数骨架见 stubs/stub.py。
应用背景
这是所有多策略 P&L 归因引擎的核心数据结构。生产环境里一个执行通道往往同时承载数十个算法(做市、统计套利、动量、篮子对冲),它们共享同一个下单路由打到同一个标的;CFO、风控官与每个算法自身的风控限额都需要分别看到每条策略的 P&L 线——把它们 net 起来会摧毁归因,让一个真正盈利的算法淹没在亏损的总账里。盯市事件在实战里要么由顶层报价 tick 触发(实时监控),要么由收盘价触发(官方 P&L 入账),同一份数据结构两种节奏都能处理——区别只在 mark 事件的出现频率。「成本基础只在开仓加仓时变」(上面规则 c)是标准的会计约定:卖掉部分多头时,你是在按当下价格实现一段已有的盈利,而不是改变剩余股票的均价;只有当你「再买入」(或者从 0 开仓)时才会重新加权 avg_cost。close-and-reverse 在跨过 0 时的处理是会计上的便利约定:旧方向被完全平掉之后,残余仓位等价于以 trade price 全新开仓,这样残余的 avg_cost 才有定义。
约束条件
- 0 ≤ len(events) ≤ 10000
- 每个 event 是一个 dict,'type' 字段取值为 {'trade', 'mark', 'ignore'} 或任意其它字符串(视作 ignore)
- trade 事件携带 strategy_id (str)、qty (int 或 float,带符号)、price (有限 float)
- mark 事件携带 mark_price (有限 float)
- 浮点比较容差 rel_tol=1e-9, abs_tol=1e-12
- qty == 0 是合法的 trade,必须是 no-op
- 若某策略首次 trade 发生在 mark 之后,那笔 mark 不会追溯生效;若它在所有 trade 之前发生,更不会让任何策略出现在输出 dict 里
样例
Case 1 · empty events list — empty dict
输入: [[]]
期望: {}
空事件流:没有任何 strategy 被见过,返回空 dict。
Case 2 · single buy trade — opens position, no realized, no mark yet
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100}]]
期望: {"S1":{"position":10,"avg_cost":100,"realized_pnl":0,"last_unrealized_pnl":0}}
单笔买入 10@100:position=10, avg_cost=100, 没有任何已实现 P&L,没有 mark 事件所以 last_unrealized_pnl=0。
Case 3 · close-and-reverse flip realizes P&L and resets cost basis on residual
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100},{"type":"trade","strategy_id":"S1","qty":-15,"price":110}]]
期望: {"S1":{"position":-5,"avg_cost":110,"realized_pnl":100,"last_unrealized_pnl":0}}
先多 10@100 后再卖 15@110:先平掉 10 实现 10*(110-100)=100 的盈利;剩下 -5 反向开仓,avg_cost 重置为 110。
Case 4 · mark snapshot computes unrealized on open position
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100},{"type":"mark","mark_price":105}]]
期望: {"S1":{"position":10,"avg_cost":100,"realized_pnl":0,"last_unrealized_pnl":50}}
持有 10@100,mark 价格 105:未实现 P&L = 10 * (105 - 100) = 50。
最近提交
还没有提交记录。
编码区
实现 solution(...)。本地运行当前支持 Python 可见样例;服务端提交会运行可见样例和隐藏测试。
默认展示公开样例。点击「运行样例」后会在这里显示实际输出;点击「提交评测」会进入隐藏测试。
Case 1 · empty events list — empty dict
输入: [[]]
期望: {}
空事件流:没有任何 strategy 被见过,返回空 dict。
Case 2 · single buy trade — opens position, no realized, no mark yet
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100}]]
期望: {"S1":{"position":10,"avg_cost":100,"realized_pnl":0,"last_unrealized_pnl":0}}
单笔买入 10@100:position=10, avg_cost=100, 没有任何已实现 P&L,没有 mark 事件所以 last_unrealized_pnl=0。
Case 3 · close-and-reverse flip realizes P&L and resets cost basis on residual
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100},{"type":"trade","strategy_id":"S1","qty":-15,"price":110}]]
期望: {"S1":{"position":-5,"avg_cost":110,"realized_pnl":100,"last_unrealized_pnl":0}}
先多 10@100 后再卖 15@110:先平掉 10 实现 10*(110-100)=100 的盈利;剩下 -5 反向开仓,avg_cost 重置为 110。
Case 4 · mark snapshot computes unrealized on open position
输入: [[{"type":"trade","strategy_id":"S1","qty":10,"price":100},{"type":"mark","mark_price":105}]]
期望: {"S1":{"position":10,"avg_cost":100,"realized_pnl":0,"last_unrealized_pnl":50}}
持有 10@100,mark 价格 105:未实现 P&L = 10 * (105 - 100) = 50。