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 )