From 5019b7d9df0e8a40701f4ca099b75348bdc42b19 Mon Sep 17 00:00:00 2001 From: PureSaber <88314620+PureSaber@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:15:43 +0800 Subject: [PATCH 1/4] Complete institutional research accounts, calibration and acceptance --- README.md | 6 + src/quant_execution/market_making.py | 270 ++++++++++++++++++ .../market_making_calibration.py | 94 ++++++ src/quant_execution/market_making_cli.py | 163 +++++++++++ src/quant_execution/research_account.py | 268 +++++++++++++++++ src/quant_execution/rules.py | 5 +- 6 files changed, 805 insertions(+), 1 deletion(-) create mode 100644 src/quant_execution/market_making.py create mode 100644 src/quant_execution/market_making_calibration.py create mode 100644 src/quant_execution/market_making_cli.py create mode 100644 src/quant_execution/research_account.py diff --git a/README.md b/README.md index 5f1b00c..e24f08c 100644 --- a/README.md +++ b/README.md @@ -232,3 +232,9 @@ the earlier materialized-path profile remain disclosed separately in The artifact-retention contract, differential matrix, benchmark definition and current gate evidence are documented in [`docs/performance-m7-streaming.md`](docs/performance-m7-streaming.md). + +## 机构研究扩展（2026-10-11） + +新增库存报价策略与命令行，复用 L2/BBO、费用、风控和精确账本。现金账户库存保持非负。 + +统一入口与合成输入用法见相邻仓库 [quant-workspace/profiles/institutional-research](../quant-workspace/profiles/institutional-research/README.md)。此增量提供离线合成验收、账户回放和独立 Studio 入口；真实市场数据与权限仍须明确提供。验收命令及边界见上述入口。 diff --git a/src/quant_execution/market_making.py b/src/quant_execution/market_making.py new file mode 100644 index 0000000..adc8796 --- /dev/null +++ b/src/quant_execution/market_making.py @@ -0,0 +1,270 @@ +"""Inventory-aware cash-equity/ETF quoting on the shared offline replay engine.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +from datetime import timedelta +from decimal import ROUND_CEILING, ROUND_FLOOR + +from quant_data_kit import ( + AssetClass, + BookDeltaEvent, + BookSnapshotEvent, + FixedPoint, + QuoteEvent, + StatusEvent, +) +from quant_data_kit.l2_replay import L2BookReconstructor +from quant_data_kit.market_events_v2 import market_event_payload + +from ._fixed import decimal, fixed +from .broker import DeterministicBroker +from .contracts import OrderIntent, OrderType, Side, TimeInForce +from .engine import DeterministicRunEngine +from .ledger import ExactAccountLedger +from .matching import L2MatchingModel, TradeBBOModel +from .rules import RuleBookRiskGate + + +@dataclass(frozen=True) +class QuotePolicy: + order_quantity: FixedPoint + max_inventory: FixedPoint + target_inventory: FixedPoint + half_spread_ticks: int = 1 + inventory_skew_ticks: int = 4 + max_quote_age_ms: int = 1000 + refresh_ms: int = 100 + cancel_latency_ms: int = 10 + + def __post_init__(self): + for name in ("order_quantity", "max_inventory"): + if not isinstance(getattr(self, name), FixedPoint) or getattr(self, name).units <= 0: + raise ValueError(f"positive FixedPoint {name} required") + if not isinstance(self.target_inventory, FixedPoint): + raise TypeError("FixedPoint target_inventory required") + if not 0 <= decimal(self.target_inventory) <= decimal(self.max_inventory): + raise ValueError("cash inventory target must lie within [0,max_inventory]") + for name in ( + "half_spread_ticks", + "inventory_skew_ticks", + "max_quote_age_ms", + "refresh_ms", + "cancel_latency_ms", + ): + low = 1 if name in {"half_spread_ticks", "max_quote_age_ms", "refresh_ms"} else 0 + if type(getattr(self, name)) is not int or getattr(self, name) < low: + raise ValueError(f"invalid integer {name}") + + +def quote_plan(spec, policy, *, bid, ask, inventory): + """Passive tick-aligned quotes, long-only with independent worst-case sizes.""" + tick, step = decimal(spec.price_tick), decimal(spec.quantity_step) + bid, ask, qty = decimal(bid), decimal(ask), decimal(inventory) + if bid <= 0 or ask <= bid or qty < 0 or qty > decimal(policy.max_inventory): + raise ValueError("invalid BBO or inventory outside policy bounds") + for amount in (policy.order_quantity, policy.max_inventory, policy.target_inventory): + if decimal(amount) % step: + raise ValueError("inventory policy must align to instrument quantity step") + center = (bid + ask) / 2 - ( + (qty - decimal(policy.target_inventory)) + / decimal(policy.max_inventory) + * policy.inventory_skew_ticks + * tick + ) + result = [] + for side, price, capacity, rounding in ( + ( + Side.BUY, + min(bid, center - policy.half_spread_ticks * tick), + decimal(policy.max_inventory) - qty, + ROUND_FLOOR, + ), + (Side.SELL, max(ask, center + policy.half_spread_ticks * tick), qty, ROUND_CEILING), + ): + price = (price / tick).to_integral_value(rounding=rounding) * tick + size = (min(decimal(policy.order_quantity), capacity) / step).to_integral_value( + rounding=ROUND_FLOOR + ) * step + if size > 0 and price > 0: + result.append( + (side, fixed(price, spec.price_tick.scale), fixed(size, spec.quantity_step.scale)) + ) + return result + + +class InventoryMarketMaker: + """Callbacks cancel/requote after the current market event has been matched. + + Cancellation latency is rounded UP to the next observed callback; old quotes + remain executable through that callback's matching phase. No time travel. + """ + + sends_live_orders = False + + def __init__(self, spec, policy, broker, gate, ledger): + if spec.asset_class not in {AssetClass.EQUITY, AssetClass.ETF}: + raise ValueError("this policy supports cash equities/ETFs only") + self.spec, self.policy = spec, policy + self.broker, self.gate, self.ledger = broker, gate, ledger + self.reset() + + def reset(self): + self._state = { + "book": L2BookReconstructor(), + "bbo": None, + "last_submit": None, + "cancel_due": None, + "decisions": [], + } + + def capture_state(self): + return deepcopy(self._state) + + def restore_state(self, state): + self._state = deepcopy(state) + + @property + def decisions(self): + return deepcopy(self._state["decisions"]) + + def on_event(self, context, event): + if event.instrument_id != self.spec.instrument_id: + return () + at, state, policy = event.available_at, self._state, self.policy + if isinstance(event, (BookSnapshotEvent, BookDeltaEvent)): + book = state["book"].apply(market_event_payload(event)) + if book.bids and book.asks: + bid, ask = book.bids[0], book.asks[0] + state["bbo"] = ( + FixedPoint(bid.price_units, bid.price_scale), + FixedPoint(ask.price_units, ask.price_scale), + event.event_time, + ) + else: + state["bbo"] = None + elif isinstance(event, QuoteEvent): + state["bbo"] = (event.bid_price, event.ask_price, event.event_time) + elif isinstance(event, StatusEvent): + # A fresh book/quote is needed after status changes, including reopens. + state["bbo"] = None + live = [ + order + for order in self.broker.open_orders + if ( + order.intent.strategy_id == context.strategy_id + and order.intent.instrument_id == event.instrument_id + ) + ] + if state["cancel_due"] is not None: + if at < state["cancel_due"]: + return () + for order in live: + self.broker.cancel( + order.order_id, + idempotency_key=f"mm-cancel:{order.order_id}:{event.event_id}", + created_at=at, + ) + self.gate.release_order(order) + live = [] + state["cancel_due"] = None + quote = state["bbo"] + fresh = quote and timedelta(0) <= at - quote[2] <= timedelta( + milliseconds=policy.max_quote_age_ms + ) + inventory = self.ledger.snapshot(at).positions.get(event.instrument_id, FixedPoint(0, 0)) + plan = ( + quote_plan(self.spec, policy, bid=quote[0], ask=quote[1], inventory=inventory) + if fresh + else [] + ) + desired = {(s, decimal(p), decimal(q)) for s, p, q in plan} + outstanding = { + ( + o.intent.side, + decimal(o.intent.limit_price), + decimal(o.intent.quantity) - decimal(o.filled_quantity), + ) + for o in live + } + if live and desired != outstanding: + state["cancel_due"] = at + timedelta(milliseconds=policy.cancel_latency_ms) + state["decisions"].append( + { + "at": at.isoformat(), + "action": "cancel_requested", + "effective_no_earlier_than": state["cancel_due"].isoformat(), + } + ) + # Even zero-latency cancels are applied at the next callback. + return () + if ( + live + or not plan + or ( + state["last_submit"] is not None + and at - state["last_submit"] < timedelta(milliseconds=policy.refresh_ms) + ) + ): + return () + state["last_submit"] = at + state["decisions"].append( + {"at": at.isoformat(), "action": "quote", "inventory": str(decimal(inventory))} + ) + return tuple( + OrderIntent( + idempotency_key=f"mm:{event.event_id}:{side.value}", + account_id=context.account_id, + strategy_id=context.strategy_id, + instrument_id=event.instrument_id, + side=side, + quantity=qty, + limit_price=price, + order_type=OrderType.LIMIT, + time_in_force=TimeInForce.DAY, + created_at=at, + ) + for side, price, qty in plan + ) + + +def replay_market_making(events, spec, policy, *, initial_cash, matching="l2", latency_ms=1): + """Return a replay engine and quote audit. Cash/fees/positions use shared ledgers.""" + if matching not in {"l2", "bbo"} or type(latency_ms) is not int or latency_ms < 1: + raise ValueError("matching must be l2/bbo; positive integer order latency required") + if not isinstance(initial_cash, FixedPoint) or initial_cash.units <= 0: + raise ValueError("positive FixedPoint cash required") + events = tuple(events) + if not events or any(e.instrument_id != spec.instrument_id for e in events): + raise ValueError("nonempty single-instrument stream required") + if matching == "l2" and not any(isinstance(e, BookSnapshotEvent) for e in events): + raise ValueError("L2 replay needs snapshots; use bbo explicitly for reduced fidelity") + if matching == "bbo" and any( + isinstance(e, (BookSnapshotEvent, BookDeltaEvent)) for e in events + ): + raise ValueError("BBO replay cannot silently ignore supplied L2 liquidity") + ledger = ExactAccountLedger( + account_id="mm-research", + base_currency=spec.settlement_currency, + instruments={spec.instrument_id: spec}, + initial_cash={spec.settlement_currency: initial_cash}, + ) + broker = DeterministicBroker() + gate = RuleBookRiskGate(instruments={spec.instrument_id: spec}, ledger=ledger) + strategy = InventoryMarketMaker(spec, policy, broker, gate, ledger) + model = L2MatchingModel if matching == "l2" else TradeBBOModel + engine = DeterministicRunEngine( + run_id="inventory-market-making", + account_id=ledger.account_id, + strategy_id="inventory-mm", + strategy=strategy, + broker=broker, + risk_gate=gate, + matching_model=model( + {spec.instrument_id: spec}, latency=timedelta(milliseconds=latency_ms) + ), + ledger=ledger, + ) + engine.replay(events, seed=0) + return engine, strategy.decisions diff --git a/src/quant_execution/market_making_calibration.py b/src/quant_execution/market_making_calibration.py new file mode 100644 index 0000000..da25799 --- /dev/null +++ b/src/quant_execution/market_making_calibration.py @@ -0,0 +1,94 @@ +"""Chronological selection of predeclared fill models from observed executions. + +Candidate predictions must be produced with the same orders and observation +window. This scores queue/latency assumptions; it never certifies exchange fills. +""" + +import argparse +import hashlib +import json +from decimal import Decimal +from pathlib import Path + +from quant_data_kit.derivatives.models import decimal, utc + + +def calibrate(payload, as_of): + if payload.get("evidence_kind") not in {"synthetic", "retrospective"}: + raise ValueError("explicit execution evidence kind required") + candidates, records = payload["candidates"], payload["records"] + if not candidates or len(records) < 10: + raise ValueError("candidate definitions and >=10 execution observations required") + rows, seen = [], set() + for row in records: + if row["order_id"] in seen or not row.get("source"): + raise ValueError("unique order IDs and source evidence required") + seen.add(row["order_id"]) + if not utc(row["submitted_at"]) < utc(row["outcome_available_at"]) <= utc(as_of): + raise ValueError( + "execution outcomes must follow submission and be known as of calibration" + ) + quantity = decimal(row["quantity"]) + if quantity <= 0 or set(row["predicted_fills"]) != set(candidates): + raise ValueError("positive order size and predictions for every candidate required") + if any( + not 0 <= decimal(value) <= quantity + for value in [row["filled_quantity"], *row["predicted_fills"].values()] + ): + raise ValueError("actual/predicted fills must be within order quantity") + rows.append(row) + rows.sort(key=lambda row: (utc(row["submitted_at"]), row["order_id"])) + boundary = utc(rows[int(len(rows) * 0.7)]["submitted_at"]) + train = [r for r in rows if utc(r["outcome_available_at"]) < boundary] + evaluation = [r for r in rows if utc(r["submitted_at"]) >= boundary] + if len(train) < 5 or len(evaluation) < 3: + raise ValueError("purged chronological fit/evaluation are too small") + + def score(group, name): + errors = [ + (decimal(r["predicted_fills"][name]) - decimal(r["filled_quantity"])) + / decimal(r["quantity"]) + for r in group + ] + return { + "n": len(errors), + "mean_squared_fill_fraction_error": str( + sum((e * e for e in errors), Decimal(0)) / len(errors) + ), + "mean_bias": str(sum(errors) / len(errors)), + } + + scores = {name: score(train, name) for name in candidates} + selected = min( + candidates, + key=lambda name: (decimal(scores[name]["mean_squared_fill_fraction_error"]), name), + ) + return { + "schema": "quant.fill-calibration/1", + "evidence_kind": payload["evidence_kind"], + "available_at": utc(as_of).isoformat(), + "selected_candidate": selected, + "selected_parameters": candidates[selected], + "training_scores": scores, + "evaluation": score(evaluation, selected), + "evaluation_order_ids": [r["order_id"] for r in evaluation], + "input_sha256": hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest(), + "limits": "Selection uses training executions only. Held-out scores are not refitted. " + "Observed fills, aligned counterfactual predictions and vendor rights must be supplied; no venue certification.", + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description="基于执行证据选择成交模拟参数") + for key in ("input", "as-of", "output"): + parser.add_argument("--" + key, required=True) + args = parser.parse_args(argv) + report = calibrate(json.loads(Path(args.input).read_text("utf-8-sig")), args.as_of) + with Path(args.output).open("x", encoding="utf-8") as file: + json.dump(report, file, ensure_ascii=False, indent=2, allow_nan=False) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/quant_execution/market_making_cli.py b/src/quant_execution/market_making_cli.py new file mode 100644 index 0000000..5c5ce82 --- /dev/null +++ b/src/quant_execution/market_making_cli.py @@ -0,0 +1,163 @@ +"""Offline JSON entry point using QDK v2 event payloads and shared accounting.""" + +import argparse +import hashlib +import json +from datetime import date, datetime +from pathlib import Path + +from quant_data_kit import ( + AggressorSide, + AssetClass, + BookAction, + BookDeltaEvent, + BookLevel, + BookSide, + BookSnapshotEvent, + FixedPoint, + InstrumentSpec, + QuoteEvent, + StatusEvent, + TradeEvent, +) + +from .market_making import QuotePolicy, replay_market_making +from .schemas import execution_payload + + +def stamp(value): + result = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if result.tzinfo is None: + raise ValueError("timezone-aware timestamp required") + return result + + +def event_from_payload(raw): + value = dict(raw) + kind = value.pop("event_type") + classes = { + "quote": QuoteEvent, + "trade": TradeEvent, + "book_snapshot": BookSnapshotEvent, + "book_delta": BookDeltaEvent, + "status": StatusEvent, + } + if kind not in classes: + raise ValueError("market-making CLI supports quote/trade/L2/status events") + for key in ("event_time", "received_at", "available_at"): + value[key] = stamp(value[key]) + value["trading_day"] = date.fromisoformat(value["trading_day"]) + for key in ("bid_price", "ask_price", "bid_quantity", "ask_quantity", "price", "quantity"): + if key in value: + value[key] = FixedPoint(**value[key]) + for side in ("bids", "asks"): + if side in value: + value[side] = tuple( + BookLevel( + price=FixedPoint(**level["price"]), + quantity=FixedPoint(**level["quantity"]), + order_count=level.get("order_count"), + ) + for level in value[side] + ) + if kind == "trade": + value["aggressor_side"] = AggressorSide(value["aggressor_side"]) + if kind == "book_delta": + value["side"], value["action"] = BookSide(value["side"]), BookAction(value["action"]) + return classes[kind](**value) + + +def evaluate(raw): + """Run deterministic research in memory, without changing inputs or files.""" + contents = json.dumps(raw, sort_keys=True, allow_nan=False).encode() + if raw.get("schema") != "quant.market-making-input/1": + raise ValueError("unsupported market-making input schema") + if raw.get("evidence_kind") not in {"synthetic", "retrospective"}: + raise ValueError("declare synthetic or retrospective inputs") + if not isinstance(raw.get("source_ref"), str) or not raw["source_ref"].strip(): + raise ValueError("source_ref is required") + spec = dict(raw["instrument"]) + spec["asset_class"] = AssetClass(spec["asset_class"]) + for key in ("price_tick", "quantity_step", "contract_multiplier"): + spec[key] = FixedPoint(**spec[key]) + for key in ("effective_from", "available_at"): + spec[key] = stamp(spec[key]) + # Deliberately keep this CLI on explicit US cash rules; other market-specific + # configurations use the Python API and their own supported rule contracts. + if spec["product_type"] not in {"us_equity", "us_etf"} or spec["settlement_currency"] != "USD": + raise ValueError("CLI v1 supports explicitly configured US cash equity/ETF") + instrument = InstrumentSpec(**spec) + policy = dict(raw["policy"]) + for key in ("order_quantity", "max_inventory", "target_inventory"): + policy[key] = FixedPoint(**policy[key]) + engine, decisions = replay_market_making( + [event_from_payload(row) for row in raw["events"]], + instrument, + QuotePolicy(**policy), + initial_cash=FixedPoint(**raw["initial_cash"]), + matching=raw.get("matching", "l2"), + latency_ms=raw.get("latency_ms", 1), + ) + artifacts = engine.artifacts + report = { + "title": "库存报价与成交模拟", + "currency": "USD", + "schema": "quant.market-making-study/1", + "evidence_kind": raw["evidence_kind"], + "input_sha256": hashlib.sha256(contents).hexdigest(), + "source_ref": raw["source_ref"], + "source_sha256": { + name: hashlib.sha256((Path(__file__).parent / name).read_bytes()).hexdigest() + for name in ("market_making.py", "market_making_cli.py", "rules.py", "matching.py") + }, + "matching": raw.get("matching", "l2"), + "config": {k: v for k, v in raw.items() if k != "events"}, + "result": execution_payload(artifacts.result), + "decisions": decisions, + "final_account": execution_payload(engine.ledger.snapshot()), + "risk_events": list(artifacts.risk_events), + "orders": [execution_payload(v) for v in artifacts.orders], + "fills": [execution_payload(v) for v in artifacts.fills], + "fees": [execution_payload(v) for v in artifacts.fees], + "ledger": [execution_payload(v) for v in artifacts.ledger_transactions], + "limits": [ + "Long-only cash inventory; no naked shorts or margin-financed market-making", + "L2 queue-ahead is a model; BBO mode lacks queue priority", + "Current event matches before cancellation/quote decisions; cancellation is event-driven", + "Fees, ticks and lot sizes are input assumptions; no exchange-priority certification", + "Ending inventory is marked, not forcibly liquidated; no live order transmission", + ], + } + return report + + +def run(input_path, output): + report = evaluate(json.loads(Path(input_path).read_bytes())) + root = Path(output) + root.mkdir(parents=True, exist_ok=False) + payload = (json.dumps(report, ensure_ascii=False, indent=2, allow_nan=False) + "\n").encode() + (root / "market-making.json").write_bytes(payload) + (root / "artifacts.json").write_text( + json.dumps({"market-making.json": hashlib.sha256(payload).hexdigest()}, indent=2), + encoding="utf-8", + ) + return { + "output": str(root), + "evidence_kind": report["evidence_kind"], + "fills": len(report["fills"]), + "ledger_sha256": report["result"]["ledger_sha256"], + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description="离线库存报价与成交模拟") + parser.add_argument("--input", required=True) + parser.add_argument("--output", required=True) + args = parser.parse_args(argv) + result = run(args.input, args.output) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/quant_execution/research_account.py b/src/quant_execution/research_account.py new file mode 100644 index 0000000..bba9da6 --- /dev/null +++ b/src/quant_execution/research_account.py @@ -0,0 +1,268 @@ +"""Borrow-gated, premium-funded research baskets on one exact shared ledger. + +Cash equities use the existing signed delivery accounting explicitly. Short sale +proceeds and a declared collateral add-on are reserved; settlement is evidenced +per basket, never assumed to grant broker buying power. No live orders. +""" + +from copy import deepcopy +from dataclasses import replace +from decimal import ROUND_HALF_EVEN, Decimal + +from quant_data_kit import AssetClass, FixedPoint, MarkPriceEvent +from quant_data_kit.derivatives.models import decimal, utc +from quant_data_kit.exceptions import ValidationError + +from .contracts import Fee, Fill, Funding, Side +from .ledger import ExactAccountLedger +from .schemas import execution_payload + + +def fp(value): + return FixedPoint.from_decimal(decimal(value), 8, rounding=ROUND_HALF_EVEN) + + +def ledger_rows(report): + """Exact decimal postings for independently checkable CSV artifacts.""" + return [ + { + "transaction_id": tx["transaction_id"], + "event_time": tx["event_time"], + "event_type": tx["event_type"], + "currency": p["currency"], + "ledger_account": p["ledger_account"], + "instrument_id": p["instrument_id"], + "amount": str(FixedPoint(**p["amount"]).to_decimal()), + } + for tx in report["ledger"] + for p in tx["postings"] + ] + + +class ResearchBasketAccount: + sends_live_orders = False + + def __init__(self, specs, initial_cash, opened_at, *, short_collateral_addon="0.5"): + self.specs = { + s.instrument_id: replace(s, metadata={**s.metadata, "research_delivery": "true"}) + if s.asset_class in {AssetClass.EQUITY, AssetClass.ETF} + else s + for s in specs + } + currencies = {s.settlement_currency for s in self.specs.values()} + if not self.specs or len(currencies) != 1 or decimal(initial_cash) <= 0: + raise ValueError("one currency, nonempty specs and positive cash required") + self.currency = next(iter(currencies)) + self.addon = decimal(short_collateral_addon) + if not 0 <= self.addon <= 5: + raise ValueError("invalid short collateral add-on") + if any( + s.asset_class not in {AssetClass.EQUITY, AssetClass.ETF, AssetClass.OPTION} + for s in self.specs.values() + ): + raise ValueError("premium-funded spot equities/ETFs and options required") + self.ledger = ExactAccountLedger( + account_id="research-basket", + base_currency=self.currency, + instruments=self.specs, + initial_cash={self.currency: fp(initial_cash)}, + opened_at=utc(opened_at), + ) + self.pending, self.fills, self.events = [], [], {} + + def quantity(self, identity): + return self.ledger.snapshot().positions.get(identity, fp(0)).to_decimal() + + def mark(self, identity, price, at, event_id): + at = utc(at) + self.ledger.mark( + MarkPriceEvent( + event_id=event_id, + instrument_id=identity, + event_time=at, + received_at=at, + available_at=at, + source="declared-research", + trading_day=at.date(), + session_id=at.date().isoformat(), + sequence=0, + price=fp(price), + ) + ) + + def _collateral(self, marks): + return sum( + ( + abs(self.quantity(i)) * decimal(marks[i]) * (1 + self.addon) + for i, s in self.specs.items() + if self.quantity(i) < 0 and s.asset_class in {AssetClass.EQUITY, AssetClass.ETF} + ), + Decimal(0), + ) + + def execute(self, legs, *, at, event_id, marks, borrow=None, settles_at=None, fee_per_unit=0): + at = utc(at) + settles = utc(settles_at) if settles_at else at + if settles < at or decimal(fee_per_unit) < 0: + raise ValueError("settlement cannot precede execution; nonnegative fees required") + legs = tuple((i, decimal(q), decimal(p)) for i, q, p in legs) + fingerprint = repr((legs, at, sorted(marks.items()), borrow, settles, str(fee_per_unit))) + if event_id in self.events: + if self.events[event_id] != fingerprint: + raise ValueError("basket ID reused with changed content") + return + if not event_id or not legs or len({i for i, _, _ in legs}) != len(legs): + raise ValueError("nonempty unique basket and ID required") + checkpoint = self.ledger.capture_state() + previous = deepcopy((self.pending, self.fills)) + try: + # Refresh valuation before calculating collateral and trading. No quote + # from outside this observed basket is inferred by the account. + for i in self.specs: + if self.quantity(i) or any(i == leg[0] for leg in legs): + self.mark(i, marks[i], at, f"{event_id}:mark:{i}") + for n, (identity, quantity, price) in enumerate(legs): + spec = self.specs[identity] + if quantity == 0 or price <= 0 or quantity % spec.quantity_step.to_decimal(): + raise ValueError("positive price, nonzero step-aligned quantity required") + if price % spec.price_tick.to_decimal(): + raise ValueError("execution price is not tick-aligned") + if not max(spec.available_at, spec.effective_from) <= at: + raise ValueError("instrument terms unavailable at execution") + old, final = self.quantity(identity), self.quantity(identity) + quantity + if final < 0 and spec.asset_class in {AssetClass.EQUITY, AssetClass.ETF}: + evidence = (borrow or {}).get(identity) + if not evidence or not ( + utc(evidence["available_at"]) <= at < utc(evidence["expires_at"]) + ): + raise ValueError("valid borrow evidence required for short inventory") + if decimal(evidence["units"]) < abs(final) or not evidence.get("source"): + raise ValueError("insufficient evidenced borrow units") + if ( + final < 0 + and spec.asset_class is AssetClass.OPTION + and "initial_margin_per_contract" not in spec.metadata + ): + raise ValueError("short option requires declared margin") + key = f"{event_id}:{n}" + self.ledger.apply( + Fill( + fill_id=key, + order_id=key, + account_id=self.ledger.account_id, + strategy_id="research-basket", + instrument_id=identity, + side=Side.BUY if quantity > 0 else Side.SELL, + quantity=fp(abs(quantity)), + price=fp(price), + event_time=at, + ) + ) + fee = fp(abs(quantity) * decimal(fee_per_unit)) + self.ledger.apply( + Fee( + fee_id=key + ":fee", + fill_id=key, + account_id=self.ledger.account_id, + amount=fee, + currency=self.currency, + event_time=at, + fee_type="commission", + ) + ) + # Only disposal of pre-owned positive units releases sale proceeds. + long_sale = min(max(old, Decimal(0)), max(-quantity, Decimal(0))) + if long_sale and settles > at: + self.pending.append( + (settles, long_sale * price * spec.contract_multiplier.to_decimal()) + ) + self.fills.append( + { + "at": at.isoformat(), + "instrument_id": identity, + "quantity": str(quantity), + "price": str(price), + "fee": str(fee.to_decimal()), + "settles_at": settles.isoformat(), + "event_id": key, + } + ) + snap = self.ledger.snapshot(at) + unsettled = sum((amount for date, amount in self.pending if date > at), Decimal(0)) + collateral = self._collateral(marks) + required = collateral + snap.initial_margin.to_decimal() + unsettled + if ( + self.ledger.cash_balance(self.currency) < required + or snap.nav.to_decimal() < snap.initial_margin.to_decimal() + ): + raise ValueError( + "insufficient cash after margin, short collateral and unsettled proceeds" + ) + self.validate() + except Exception as exc: + self.ledger.restore_state(checkpoint) + self.pending, self.fills = previous + if isinstance(exc, ValidationError): + raise ValueError(str(exc)) from exc # noqa: TRY004 - normalize ledger validation + raise + self.events[event_id] = fingerprint + + def charge(self, amount, at, event_id, kind="funding_and_borrow"): + amount = fp(amount) + if amount.units < 0: + raise ValueError("nonnegative costs required") + self.ledger.apply( + Funding( + funding_id=kind + ":" + event_id, + instrument_id=next(iter(self.specs)), + account_id=self.ledger.account_id, + amount=fp(-amount.to_decimal()), + currency=self.currency, + event_time=utc(at), + ) + ) + + def check_risk(self, marks, at, borrow=None): + """Validate carried inventory even when there is no new order.""" + at = utc(at) + for identity, spec in self.specs.items(): + if self.quantity(identity) >= 0 or spec.asset_class is AssetClass.OPTION: + continue + evidence = (borrow or {}).get(identity) + if not evidence or not utc(evidence["available_at"]) <= at < utc( + evidence["expires_at"] + ): + raise ValueError("carried short requires unexpired borrow evidence") + if decimal(evidence["units"]) < abs(self.quantity(identity)) or not evidence.get( + "source" + ): + raise ValueError("carried short exceeds evidenced borrow") + snap = self.ledger.snapshot(at) + pending = sum((value for date, value in self.pending if date > at), Decimal(0)) + required = self._collateral(marks) + snap.initial_margin.to_decimal() + pending + if ( + self.ledger.cash_balance(self.currency) < required + or snap.nav.to_decimal() < snap.initial_margin.to_decimal() + ): + raise ValueError("insufficient carried collateral or margin") + self.validate() + + def validate(self): + for tx in self.ledger.transactions: + totals = {} + for posting in tx.postings: + totals[posting.currency] = ( + totals.get(posting.currency, Decimal(0)) + posting.amount.to_decimal() + ) + if any(totals.values()): + raise AssertionError("unbalanced research transaction") + + def report(self): + self.validate() + return { + "final_account": execution_payload(self.ledger.snapshot()), + "fills": self.fills, + "ledger": [execution_payload(tx) for tx in self.ledger.transactions], + "ledger_balanced": True, + "sends_live_orders": False, + } diff --git a/src/quant_execution/rules.py b/src/quant_execution/rules.py index 7e4c75c..76f506a 100644 --- a/src/quant_execution/rules.py +++ b/src/quant_execution/rules.py @@ -913,7 +913,10 @@ def _check_reservations( if reserved_instrument == instrument_id and key != intent.idempotency_key ) available_position = abs(_value(snapshot.positions.get(instrument_id))) - if spec.asset_class in {AssetClass.EQUITY, AssetClass.ETF}: + if spec.asset_class in { + AssetClass.EQUITY, + AssetClass.ETF, + } and spec.product_type.lower() not in {"us_equity", "us_etf"}: available_position -= self.ledger.acquired_today( instrument_id, state.event.trading_day ) From adbe678f6603aa46aaa21d53de04ce8ab7342285 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Sun, 11 Oct 2026 11:21:14 +0800 Subject: [PATCH 2/4] Add exact offline stock and option lifecycle accounting --- docs/research-lifecycle.md | 23 ++ src/quant_execution/research_lifecycle.py | 397 ++++++++++++++++++++++ tests/test_research_lifecycle.py | 223 ++++++++++++ 3 files changed, 643 insertions(+) create mode 100644 docs/research-lifecycle.md create mode 100644 src/quant_execution/research_lifecycle.py create mode 100644 tests/test_research_lifecycle.py diff --git a/docs/research-lifecycle.md b/docs/research-lifecycle.md new file mode 100644 index 0000000..50c4a09 --- /dev/null +++ b/docs/research-lifecycle.md @@ -0,0 +1,23 @@ +# ResearchLifecycle + +Additive offline stock/vanilla-option account, reusing existing FixedPoint, Posting, +LedgerTransaction and execution_payload contracts. No live transport and no schema +changes. Price/quantity precision is eight decimals; every transaction balances. + +Supported events: deposit, locate, fill, mark, accrue, dividend_ex, dividend_pay, +split, margin and exercise. Fills can reverse signed positions. A valid dated +borrow locate is mandatory for short stock. Apply financing before each state +change inside an interval; quant-options lifecycle does this automatically. +Unchanged event IDs are idempotent; conflicts reject; failures roll back atomically. + +Cash/physical exercise settles full positions at expiry using the exact expiry +fixing. Early exercise requires American contracts. Partial early exercise +invalidates the remaining option mark, since exercise intrinsic is not its fair +value. Pure splits require explicit value-preserving option adjustments. Margin +segregation never becomes a PnL expense; shortfalls are reported for funding or +liquidation, without invented forced fills. + +`snapshot()` reports missing/stale marks and excludes no held position silently. +`journal()` uses the existing public exact ledger schema, with quantity deltas. +`tests/test_research_lifecycle.py` validates ledger reconciliation, reversals, +assignment, rollback, funding, dividends, splits and margin. diff --git a/src/quant_execution/research_lifecycle.py b/src/quant_execution/research_lifecycle.py new file mode 100644 index 0000000..3f2fad9 --- /dev/null +++ b/src/quant_execution/research_lifecycle.py @@ -0,0 +1,397 @@ +"""Exact journal for declared stock/vanilla-option research lifecycles. + +Offline event replay only. Signed stock positions require dated borrow locates; +margin is collateral segregation, not an expense. Existing public journal types +are reused so each transaction balances per currency without schema changes. +""" + +import copy +import hashlib +import json +from decimal import ROUND_HALF_EVEN, Decimal, localcontext + +from quant_data_kit import FixedPoint +from quant_data_kit.derivatives.models import decimal, utc + +from .contracts import LedgerEventType, LedgerTransaction, Posting +from .schemas import execution_payload + + +def amount(value): + return FixedPoint.from_decimal(decimal(value), 8, rounding=ROUND_HALF_EVEN) + + +class ResearchLifecycle: + sends_live_orders = False + + def __init__(self, currency, instruments): + if not isinstance(currency, str) or len(currency) != 3 or not currency.isupper(): + raise ValueError("three-letter currency required") + self.currency = currency + self.instruments = copy.deepcopy(instruments) + if not instruments: + raise ValueError("instrument declarations required") + for spec in self.instruments.values(): + required = {"kind", "currency", "multiplier", "known_at"} + extra = {"underlying", "right", "strike", "expiry", "style", "settlement"} + if not required <= set(spec) or set(spec) - required - extra: + raise ValueError("invalid instrument specification") + if spec["currency"] != currency or spec["kind"] not in {"stock", "option"}: + raise ValueError("single currency stock/option portfolio required") + if decimal(spec["multiplier"]) <= 0: + raise ValueError("positive multiplier required") + utc(spec["known_at"]) + if spec["kind"] == "stock": + if decimal(spec["multiplier"]) != 1: + raise ValueError("stock multiplier must be one") + else: + if not extra <= set(spec) or spec["right"] not in {"call", "put"}: + raise ValueError("complete option contract required") + if spec["style"] not in {"american", "european"} or spec["settlement"] not in { + "cash", + "physical", + }: + raise ValueError("invalid option style/settlement") + if decimal(spec["strike"]) <= 0: + raise ValueError("positive strike required") + utc(spec["expiry"]) + underlying = instruments.get(spec["underlying"]) + if underlying is None or underlying["kind"] != "stock": + raise ValueError("declared underlying stock required") + self.cash = Decimal(0) + self.collateral = Decimal(0) + self.required_margin = Decimal(0) + self.positions = dict.fromkeys(instruments, Decimal(0)) + self.costs = dict.fromkeys(instruments, Decimal(0)) + self.marks = {} + self.locates = {} + self.dividends = {} + self.pending = {} + self.closed_options = set() + self.accrued_to = None + self.at = None + self.ids = {} + self.transactions = [] + self.margin_calls = [] + + def _post(self, event, postings, typ=LedgerEventType.SETTLEMENT, quantity_deltas=None): + postings = tuple( + Posting( + ledger_account=a, + currency=self.currency, + amount=amount(v), + instrument_id=i, + quantity_delta=amount(quantity_deltas[i]) + if quantity_deltas and a == "assets:position_cost" and i in quantity_deltas + else None, + ) + for a, v, i in postings + ) + transaction = LedgerTransaction( + transaction_id=event["id"] + f":{len(self.transactions)}", + idempotency_key=event["id"] + f":{len(self.transactions)}", + event_time=utc(event["at"]), + event_type=typ, + reference_id=event["id"], + postings=postings, + ) + self.transactions.append(transaction) + + def _trade(self, event, instrument, quantity, price, fee=0, *, settlement=False): + spec = self.instruments[instrument] + at = utc(event["at"]) + q, p, fee = decimal(quantity), decimal(price), decimal(fee) + if any(amount(v).to_decimal() != v for v in (q, p)): + raise ValueError("price and quantity support at most eight decimal places") + if q == 0 or p < 0 or fee < 0 or spec["known_at"] and utc(spec["known_at"]) > at: + raise ValueError("nonzero trade and available rules/nonnegative price/cost required") + if spec["kind"] == "option" and ( + q != int(q) or (not settlement and at >= utc(spec["expiry"])) + ): + raise ValueError("integer unexpired option contracts required") + if instrument in self.closed_options and not settlement: + raise ValueError("option already settled") + old, old_cost = self.positions[instrument], self.costs[instrument] + new = old + q + if spec["kind"] == "stock" and new < 0: + locate = self.locates.get(instrument) + if ( + locate is None + or not utc(locate["available_at"]) <= at <= utc(locate["expires_at"]) + or -new > decimal(locate["quantity"]) + ): + raise ValueError("short stock exceeds available unexpired borrow locate") + multiplier = decimal(spec["multiplier"]) + closed = min(abs(old), abs(q)) if old * q < 0 else Decimal(0) + removed_cost = amount(old_cost * closed / abs(old)).to_decimal() if closed else Decimal(0) + opened = abs(q) - closed + added_cost = amount(opened * (1 if q > 0 else -1) * p * multiplier).to_decimal() + cost_delta = added_cost - removed_cost + cash_delta = amount(-q * p * multiplier).to_decimal() + fee = amount(fee).to_decimal() + self._post( + event, + [ + ("assets:cash", cash_delta - fee, None), + ("assets:position_cost", cost_delta, instrument), + ("income:realized_pnl", -(cash_delta + cost_delta), instrument), + ("expenses:commission", fee, instrument), + ], + LedgerEventType.FILL, + quantity_deltas={instrument: q}, + ) + self.cash += cash_delta - fee + self.positions[instrument] = new + self.costs[instrument] += cost_delta + + def apply(self, event): + fingerprint = hashlib.sha256( + json.dumps(event, sort_keys=True, allow_nan=False).encode() + ).hexdigest() + if not event.get("id") or not isinstance(event["id"], str): + raise ValueError("event id required") + if event["id"] in self.ids: + if self.ids[event["id"]] != fingerprint: + raise ValueError("idempotency conflict") + return self.snapshot() + at = utc(event["at"]) + if self.at is not None and at < self.at: + raise ValueError("events must be chronological") + if utc(event["available_at"]) > at: + raise ValueError("event not available at processing time") + before = copy.deepcopy(self.__dict__) + try: + with localcontext() as context: + context.prec = 60 + self._apply(event) + self.at = at + self.ids[event["id"]] = fingerprint + return self.snapshot() + except Exception: + self.__dict__.clear() + self.__dict__.update(before) + raise + + def _apply(self, e): + kind, at = e["type"], utc(e["at"]) + if kind == "deposit": + value = amount(e["amount"]).to_decimal() + if value <= 0: + raise ValueError("positive capital deposit required") + self._post(e, [("assets:cash", value, None), ("equity:capital", -value, None)]) + self.cash += value + elif kind == "locate": + instrument = e["instrument"] + if ( + self.instruments[instrument]["kind"] != "stock" + or decimal(e["quantity"]) < 0 + or utc(e["expires_at"]) < at + ): + raise ValueError("valid stock locate required") + if decimal(e["quantity"]) < max(0, -self.positions[instrument]): + raise ValueError("locate recall requires covering outstanding borrow first") + self.locates[instrument] = dict(e) + elif kind == "fill": + self._trade(e, e["instrument"], e["quantity"], e["price"], e.get("fee", 0)) + elif kind == "mark": + for instrument, p in e["prices"].items(): + if instrument not in self.instruments or decimal(p) < 0: + raise ValueError("nonnegative declared instrument marks required") + self.marks[instrument] = (amount(p).to_decimal(), at) + elif kind == "accrue": + start = utc(e["start"]) + if start >= at or (self.accrued_to is not None and start != self.accrued_to): + raise ValueError("contiguous nonoverlapping funding intervals required") + # Caller supplies previous interval's balance via immutable state: any + # intervening cash/position change requires a boundary accrual event. + if self.at and start < self.at: + raise ValueError("accrue before applying cash/position changes inside interval") + days = Decimal(str((at - start).total_seconds())) / Decimal(86400) + debit, credit = decimal(e["debit_rate"]), decimal(e["credit_rate"]) + if abs(debit) > 1 or abs(credit) > 1: + raise ValueError("annual funding rate outside [-1,1]") + interest = amount( + self.cash * (debit if self.cash < 0 else credit) * days / 365 + ).to_decimal() + borrowing = Decimal(0) + for instrument, q in self.positions.items(): + if q < 0 and self.instruments[instrument]["kind"] == "stock": + if instrument not in e["borrow_rates"] or instrument not in self.marks: + raise ValueError("borrow rate and mark required for each short stock") + r = decimal(e["borrow_rates"][instrument]) + if not 0 <= r <= 5 or utc(self.locates[instrument]["expires_at"]) < at: + raise ValueError("borrow rate/locate expiry invalid during accrual") + borrowing += amount( + -q * self.marks[instrument][0] * r * days / 365 + ).to_decimal() + self._post( + e, + [ + ("assets:cash", interest - borrowing, None), + ("income:interest", -interest, None), + ("expenses:borrow", borrowing, None), + ], + LedgerEventType.FUNDING, + ) + self.cash += interest - borrowing + self.accrued_to = at + elif kind == "dividend_ex": + instrument, div_id = e["instrument"], e["dividend_id"] + if self.instruments[instrument]["kind"] != "stock" or div_id in self.dividends: + raise ValueError("distinct stock dividend required") + value = amount(self.positions[instrument] * decimal(e["amount_per_share"])).to_decimal() + if decimal(e["amount_per_share"]) < 0 or utc(e["pay_at"]) < at: + raise ValueError("nonnegative dividend and future payment required") + self.dividends[div_id] = (value, utc(e["pay_at"])) + self.pending[div_id] = value + self._post( + e, + [ + ("assets:dividend_receivable", value, instrument), + ("income:dividend", -value, instrument), + ], + LedgerEventType.CORPORATE_ACTION, + ) + elif kind == "dividend_pay": + div_id = e["dividend_id"] + if div_id not in self.pending or at < self.dividends[div_id][1]: + raise ValueError("unpaid dividend at/after payment date required") + value = self.pending.pop(div_id) + self._post( + e, + [("assets:cash", value, None), ("assets:dividend_receivable", -value, None)], + LedgerEventType.CORPORATE_ACTION, + ) + self.cash += value + elif kind == "split": + instrument = e["instrument"] + ratio = decimal(e["ratio"]) + if self.instruments[instrument]["kind"] != "stock" or ratio <= 0: + raise ValueError("positive stock split ratio required") + affected = {i for i, s in self.instruments.items() if s.get("underlying") == instrument} + if set(e["option_adjustments"]) != affected: + raise ValueError("explicit exchange adjustments required for every related option") + old_quantity = self.positions[instrument] + new_quantity = old_quantity * ratio + if amount(new_quantity).to_decimal() != new_quantity: + raise ValueError("fractional share precision requires cash-in-lieu handling") + self.positions[instrument] = new_quantity + if instrument in self.marks: + p, observed = self.marks[instrument] + self.marks[instrument] = (amount(p / ratio).to_decimal(), observed) + if instrument in self.locates: + self.locates[instrument]["quantity"] = str( + decimal(self.locates[instrument]["quantity"]) * ratio + ) + for i in affected: + spec, adjustment = self.instruments[i], e["option_adjustments"][i] + multiplier, strike = ( + decimal(adjustment["multiplier"]), + decimal(adjustment["strike"]), + ) + if multiplier != decimal(spec["multiplier"]) * ratio or strike * ratio != decimal( + spec["strike"] + ): + raise ValueError("only value-preserving pure splits supported by this event") + spec.update(multiplier=str(multiplier), strike=str(strike), known_at=e["at"]) + if i in self.marks: + p, observed = self.marks[i] + self.marks[i] = (amount(p / ratio).to_decimal(), observed) + self._post( + e, + [("assets:position_cost", 0, instrument), ("equity:split", 0, instrument)], + LedgerEventType.CORPORATE_ACTION, + {instrument: new_quantity - old_quantity}, + ) + elif kind == "margin": + requirement = amount(e["requirement"]).to_decimal() + if requirement < 0 or not e.get("model"): + raise ValueError("nonnegative margin and declared model required") + change = requirement - self.collateral + transfer = min(change, max(self.cash, Decimal(0))) if change > 0 else change + self._post(e, [("assets:cash", -transfer, None), ("assets:collateral", transfer, None)]) + self.cash -= transfer + self.collateral += transfer + self.required_margin = requirement + if requirement > self.collateral: + self.margin_calls.append( + { + "at": at.isoformat(), + "shortfall": str(requirement - self.collateral), + "model": e["model"], + "status": "requires_funding_or_liquidation", + } + ) + elif kind == "exercise": + instrument = e["instrument"] + spec = self.instruments[instrument] + if spec["kind"] != "option" or instrument in self.closed_options: + raise ValueError("open option required for exercise/expiry") + if at < utc(spec["expiry"]) and spec["style"] != "american": + raise ValueError("European option cannot exercise before expiry") + if utc(e["fixing_at"]) > at or utc(e["fixing_available_at"]) > at: + raise ValueError("settlement fixing not available") + if at >= utc(spec["expiry"]) and utc(e["fixing_at"]) != utc(spec["expiry"]): + raise ValueError("exact expiry fixing required") + spot, strike = decimal(e["fixing"]), decimal(spec["strike"]) + if spot <= 0: + raise ValueError("positive settlement fixing required") + held = self.positions[instrument] + quantity = decimal(e.get("contracts", abs(held))) + if quantity < 0 or quantity != int(quantity) or quantity > abs(held): + raise ValueError("exercise quantity exceeds position") + signed = quantity * (1 if held > 0 else -1) + right_sign = 1 if spec["right"] == "call" else -1 + intrinsic = max(right_sign * (spot - strike), Decimal(0)) + if signed: + self._trade(e, instrument, -signed, intrinsic, settlement=True) + if spec["settlement"] == "physical" and intrinsic > 0: + # Closing option for intrinsic then acquiring/delivering shares + # at fixing is exactly equivalent to paying/receiving strike. + shares = signed * decimal(spec["multiplier"]) * right_sign + self._trade(e, spec["underlying"], shares, spot, settlement=True) + if at >= utc(spec["expiry"]): + if self.positions[instrument]: + raise ValueError("expiry must settle entire outstanding position") + self.closed_options.add(instrument) + if at >= utc(spec["expiry"]) or not self.positions[instrument]: + self.marks[instrument] = (Decimal(0), at) + else: + # Intrinsic is the exercise payoff, not the fair value of the + # remaining American contracts. Require a new observed mark. + self.marks.pop(instrument, None) + self.marks[spec["underlying"]] = (spot, utc(e["fixing_at"])) + else: + raise ValueError(f"unsupported lifecycle event: {kind}") + + def snapshot(self, *, max_mark_age_seconds=None): + nav = self.cash + self.collateral + sum(self.pending.values()) + unpriced = [] + for instrument, q in self.positions.items(): + if not q: + continue + mark = self.marks.get(instrument) + if mark is None or ( + max_mark_age_seconds is not None + and self.at is not None + and (self.at - mark[1]).total_seconds() > max_mark_age_seconds + ): + unpriced.append(instrument) + else: + nav += q * mark[0] * decimal(self.instruments[instrument]["multiplier"]) + return { + "currency": self.currency, + "at": self.at.isoformat() if self.at else None, + "cash": str(self.cash), + "collateral": str(self.collateral), + "required_margin": str(self.required_margin), + "margin_shortfall": str(max(0, self.required_margin - self.collateral)), + "positions": {k: str(v) for k, v in self.positions.items()}, + "nav": str(amount(nav).to_decimal()) if not unpriced else None, + "unpriced": unpriced, + "dividend_receivables": str(sum(self.pending.values())), + "transactions": len(self.transactions), + } + + def journal(self): + return [execution_payload(t) for t in self.transactions] diff --git a/tests/test_research_lifecycle.py b/tests/test_research_lifecycle.py new file mode 100644 index 0000000..d256c32 --- /dev/null +++ b/tests/test_research_lifecycle.py @@ -0,0 +1,223 @@ +from copy import deepcopy +from decimal import Decimal + +import pytest + +from quant_execution.research_lifecycle import ResearchLifecycle + +AT = "2026-01-02T12:00:00Z" +EXPIRY = "2026-02-02T12:00:00Z" + + +def account(*, settlement="cash", style="american", right="call"): + specs = { + "S": {"kind": "stock", "currency": "USD", "multiplier": "1", "known_at": AT}, + "O": { + "kind": "option", + "currency": "USD", + "multiplier": "100", + "known_at": AT, + "underlying": "S", + "right": right, + "strike": "100", + "expiry": EXPIRY, + "style": style, + "settlement": settlement, + }, + } + return ResearchLifecycle("USD", specs) + + +def event(a, key, kind, **fields): + at = fields.pop("at", AT) + e = dict(id=key, type=kind, at=at, available_at=at, **fields) + a.apply(e) + return e + + +def balanced(a): + balances = {} + quantities = {} + for t in a.journal(): + total = Decimal(0) + for p in t["postings"]: + value = Decimal(p["amount"]["units"]) / (10 ** p["amount"]["scale"]) + total += value + balances[p["ledger_account"]] = balances.get(p["ledger_account"], Decimal(0)) + value + q = p.get("quantity_delta") + if q is not None: + quantities[p["instrument_id"]] = quantities.get( + p["instrument_id"], Decimal(0) + ) + Decimal(q["units"]) / (10 ** q["scale"]) + assert total == 0 + assert balances.get("assets:cash", 0) == a.cash + assert balances.get("assets:collateral", 0) == a.collateral + assert balances.get("assets:position_cost", 0) == sum(a.costs.values()) + for i, q in quantities.items(): + assert q == a.positions[i] + + +def test_short_reversal_realized_pnl_and_exact_journal(): + a = account() + event(a, "deposit", "deposit", amount="10000") + event(a, "locate", "locate", instrument="S", quantity="100", expires_at=EXPIRY) + event(a, "sell", "fill", instrument="S", quantity="-10", price="100", fee="1") + event(a, "reverse", "fill", instrument="S", quantity="15", price="90", fee="1") + event(a, "mark", "mark", prices={"S": "90"}) + assert a.cash == Decimal(9648) + assert a.costs["S"] == 450 + assert a.snapshot()["nav"] == "10098.00000000" + balanced(a) + + +@pytest.mark.parametrize( + "settlement,expected_cash,expected_stock", [("cash", "10750", "0"), ("physical", "-250", "100")] +) +def test_expiry_cash_or_physical_matches_economic_nav(settlement, expected_cash, expected_stock): + a = account(settlement=settlement) + event(a, "d", "deposit", amount="10000") + event(a, "buy", "fill", instrument="O", quantity="1", price="2.5") + e = event( + a, + "expiry", + "exercise", + instrument="O", + fixing="110", + fixing_at=EXPIRY, + fixing_available_at=EXPIRY, + at=EXPIRY, + ) + assert a.cash == Decimal(expected_cash) + assert a.positions["S"] == Decimal(expected_stock) + assert a.positions["O"] == 0 + assert Decimal(a.snapshot()["nav"]) == 10750 + prior = a.journal() + a.apply(e) + assert a.journal() == prior + with pytest.raises(ValueError, match="idempotency"): + a.apply(dict(e, fixing="111")) + balanced(a) + + +def test_short_assignment_requires_borrow_and_failure_rolls_back(): + a = account(settlement="physical") + event(a, "sell", "fill", instrument="O", quantity="-1", price="3") + before = deepcopy(a.__dict__) + with pytest.raises(ValueError, match="borrow locate"): + event( + a, + "expiry", + "exercise", + instrument="O", + fixing="110", + fixing_at=EXPIRY, + fixing_available_at=EXPIRY, + at=EXPIRY, + ) + assert a.__dict__ == before + event(a, "locate", "locate", instrument="S", quantity="100", expires_at=EXPIRY) + event( + a, + "expire", + "exercise", + instrument="O", + fixing="110", + fixing_at=EXPIRY, + fixing_available_at=EXPIRY, + at=EXPIRY, + ) + assert a.positions["S"] == -100 + assert Decimal(a.snapshot()["nav"]) == -700 + balanced(a) + + +def test_finance_dividend_ex_date_entitlement_and_margin_segregation(): + a = account() + event(a, "d", "deposit", amount="10000") + event(a, "loc", "locate", instrument="S", quantity="100", expires_at=EXPIRY) + event(a, "s", "fill", instrument="S", quantity="-10", price="100") + event(a, "m", "mark", prices={"S": "100"}) + event(a, "margin", "margin", requirement="2000", model="declared stress") + assert Decimal(a.snapshot()["nav"]) == 10000 + end = "2026-01-03T12:00:00Z" + event( + a, + "fund", + "accrue", + start=AT, + at=end, + credit_rate="0.0365", + debit_rate="0.073", + borrow_rates={"S": "0.365"}, + ) + assert a.cash == Decimal("8999.90") # 0.90 interest - 1 borrow + event( + a, + "ex", + "dividend_ex", + at=end, + instrument="S", + dividend_id="div", + amount_per_share="2", + pay_at=end, + ) + event(a, "cover", "fill", at=end, instrument="S", quantity="10", price="98") + event(a, "pay", "dividend_pay", at=end, dividend_id="div") + event(a, "release", "margin", at=end, requirement="0", model="flat") + assert Decimal(a.snapshot()["nav"]) == Decimal("9999.9") + balanced(a) + + +def test_split_preserves_nav_and_changes_deliverable(): + a = account() + event(a, "d", "deposit", amount="20000") + event(a, "s", "fill", instrument="S", quantity="10", price="100") + event(a, "o", "fill", instrument="O", quantity="1", price="4") + event(a, "m", "mark", prices={"S": "100", "O": "4"}) + before = a.snapshot()["nav"] + event( + a, + "split", + "split", + instrument="S", + ratio="2", + option_adjustments={"O": {"multiplier": "200", "strike": "50"}}, + ) + assert a.positions["S"] == 20 + assert a.snapshot()["nav"] == before + balanced(a) + + +def test_margin_shortfall_stale_marks_and_future_events_reject(): + a = account() + event(a, "d", "deposit", amount="100") + event(a, "m", "margin", requirement="200", model="stress") + assert a.snapshot()["margin_shortfall"] == "100.00000000" + assert len(a.margin_calls) == 1 + before = deepcopy(a.__dict__) + with pytest.raises(ValueError, match="not available"): + a.apply({"id": "x", "type": "deposit", "amount": 100, "at": AT, "available_at": EXPIRY}) + assert a.__dict__ == before + with pytest.raises(ValueError, match="borrow"): + event(a, "short", "fill", instrument="S", quantity=-1, price=100) + balanced(a) + + +def test_european_early_and_wrong_expiry_fixing_rejected(): + a = account(style="european") + event(a, "b", "fill", instrument="O", quantity=1, price=2) + with pytest.raises(ValueError, match="before expiry"): + event( + a, "early", "exercise", instrument="O", fixing=110, fixing_at=AT, fixing_available_at=AT + ) + with pytest.raises(ValueError, match="exact expiry"): + event( + a, + "late", + "exercise", + at=EXPIRY, + instrument="O", + fixing=110, + fixing_at=AT, + fixing_available_at=AT, + ) From 9673f3426cbd71906a1baa401e400df5e3b3b338 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Sun, 11 Oct 2026 11:30:05 +0800 Subject: [PATCH 3/4] Pin combined data contract for complete research accounts --- pyproject.toml | 2 +- requirements.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 5920541..a2eab38 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ requires-python = ">=3.10" dependencies = [ "pyarrow>=14.0", "jsonschema>=4.20", - "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@ab17ee7b8c417aa72a9707106f5455beba965b76", + "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@3c3fd3845d37a9190d35a77172c221805f02be2b", ] [project.optional-dependencies] diff --git a/requirements.lock b/requirements.lock index 382e96e..2e84c44 100644 --- a/requirements.lock +++ b/requirements.lock @@ -69,7 +69,7 @@ pytz==2026.3.post1 # via pandas pyyaml==6.0.3 # via quant-data-kit -quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@ab17ee7b8c417aa72a9707106f5455beba965b76 +quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@3c3fd3845d37a9190d35a77172c221805f02be2b # via quant-execution (pyproject.toml) referencing==0.37.0 # via From 44c85cc7865a47d72b4bae2ed4be16ee6238b345 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Sun, 11 Oct 2026 11:43:25 +0800 Subject: [PATCH 4/4] Freeze integrated research dependencies including Windows event fix --- pyproject.toml | 2 +- requirements.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a2eab38..b63d5dc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ requires-python = ">=3.10" dependencies = [ "pyarrow>=14.0", "jsonschema>=4.20", - "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@3c3fd3845d37a9190d35a77172c221805f02be2b", + "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@f49f1cc306a4a27a2734e21c89ec5b68f7d7384d", ] [project.optional-dependencies] diff --git a/requirements.lock b/requirements.lock index 2e84c44..7ec269a 100644 --- a/requirements.lock +++ b/requirements.lock @@ -69,7 +69,7 @@ pytz==2026.3.post1 # via pandas pyyaml==6.0.3 # via quant-data-kit -quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@3c3fd3845d37a9190d35a77172c221805f02be2b +quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@f49f1cc306a4a27a2734e21c89ec5b68f7d7384d # via quant-execution (pyproject.toml) referencing==0.37.0 # via