Trading Engine integration¶
Import the public surface from persistra.integrations.trading_engine. Trading Engine is a
separate executable and the authority for execution semantics. Persistra supports only the
current v1 contract.
Shared models¶
persistra.integrations.trading_engine.model
¶
Shared typed values for the current Trading Engine v1 contract.
TRADING_ENGINE_CONTRACT_VERSION: Final = '1'
module-attribute
¶
_STRATEGY_PROTOCOL_MAX_MESSAGE_BYTES: Final = 1048576
module-attribute
¶
_TRADING_ENGINE_DIAGNOSTIC_PHASES: Final = {'cli', 'input', 'validation', 'replay', 'reducer', 'strategy', 'artifact'}
module-attribute
¶
MissingBarVolumePolicy = Literal['reject', 'zero_impact']
¶
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
EngineResourceLimits
dataclass
¶
Versioned inclusive resource ceilings advertised by an engine.
EngineCapabilities
dataclass
¶
Machine-readable compatibility surface advertised by an engine executable.
TradingEngineDiagnosticContext
dataclass
¶
Typed location and causality fields from one engine diagnostic.
TradingEngineDiagnosticCause
dataclass
¶
Sanitized exception cause supplied by Trading Engine.
TradingEngineDiagnostic
dataclass
¶
Version-1 machine-readable failure reported by Trading Engine.
StrategyResponseEvidence
dataclass
¶
Bounded raw response bytes retained after protocol rejection.
StrategyResponseRejection
dataclass
¶
Version-1 diagnostic record ending one failed strategy transcript.
ExecutionInstrument
dataclass
¶
Executable metadata supplied separately from research identity.
ConservativeBarExecutionPolicy
dataclass
¶
Current next-open or adverse-touch completed-bar configuration.
TradingEngineProcessError
dataclass
¶
Bases: RuntimeError
A validation or replay subprocess failed.
decimal_value(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return a finite decimal with at most six fractional places.
execution_quantity(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return an exact signed quantity with six-decimal precision.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
_freeze_mapping(value: Mapping[object, object]) -> Mapping[str, Any]
¶
_freeze_value(value: object) -> object
¶
_thaw_mapping(value: Mapping[str, object]) -> dict[str, object]
¶
_thaw_value(value: object) -> object
¶
Contract schemas¶
persistra.integrations.trading_engine.contracts
¶
Schema-backed Trading Engine contract validation.
_SCHEMA_FILES = ('scenario.schema.json', 'scenario-stream.schema.json', 'journal.schema.json')
module-attribute
¶
_EXECUTION_PRICE_COLUMNS = ('engine_sequence', 'order_id', 'instrument_id', 'side', 'reference_price', 'spread_adjustment', 'impact_adjustment', 'final_price')
module-attribute
¶
TradingEngineContractError
¶
Bases: ValueError
A structural contract failure with artifact and schema-path context.
SchemaReplayResult
dataclass
¶
Schema-verified replay envelope and model-specific execution-price evidence.
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
freeze_payload(value: object) -> object
¶
Recursively freeze a validated JSON payload.
iter_json_records(path: str | Path) -> Iterator[tuple[int, dict[str, object]]]
¶
Yield strict JSON Lines records without retaining the file text or line list.
_check_references(schema: Mapping[str, object], registry: Registry[Any], *, name: str) -> None
¶
_string_values_for_key(value: object, key: str) -> set[str]
¶
_execution_models(value: object) -> set[str]
¶
_models_in_execution(value: Mapping[object, object]) -> set[str]
¶
Initial portfolio state¶
persistra.integrations.trading_engine.initial_state
¶
Typed Trading Engine initial portfolio scenarios and reconciliation.
INITIAL_STATE_CONTRACT_VERSION: Final = '1'
module-attribute
¶
_MICRO = Decimal('0.000001')
module-attribute
¶
SchemaReplayResult
dataclass
¶
Schema-verified replay envelope and model-specific execution-price evidence.
TradingEngineContractError
¶
Bases: ValueError
A structural contract failure with artifact and schema-path context.
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
ExecutionInstrument
dataclass
¶
Executable metadata supplied separately from research identity.
InstrumentRiskPolicy
dataclass
¶
Approved risk limits for exactly one catalog instrument.
RiskFinancingRiskPolicy
dataclass
¶
Aggregate and per-instrument v1 risk policy.
RiskGroup
dataclass
¶
Versioned, explicitly sourced group membership and limits.
RiskGroupLimits
dataclass
¶
Optional v1 exposure ceilings for a named risk group.
InitialCashBalance
dataclass
¶
One signed native-currency cash ledger at replay start.
InitialPosition
dataclass
¶
One signed opening quantity with complete accounting attribution.
InitialMark
dataclass
¶
One positive opening price for an initial position.
InitialFxRate
dataclass
¶
One positive currency-to-base opening rate.
InitialPortfolioState
dataclass
¶
InitialStateScenario
dataclass
¶
One complete schema-v1 scenario containing a typed initial portfolio.
to_dict() -> dict[str, object]
¶
Return the exact complete scenario v1 document.
InitialStateReconciliation
dataclass
¶
Verified schema replay with reconciled opening audit evidence.
to_dict() -> dict[str, object]
¶
Return stable manifest-ready reconciliation evidence.
freeze_portable_mapping(value: Mapping[str, Any], *, name: str, redact_api_keys: bool = False) -> Mapping[str, Any]
¶
Copy and recursively freeze one portable JSON mapping.
thaw_portable_mapping(value: Mapping[str, Any]) -> dict[str, Any]
¶
Return a mutable JSON-compatible copy of one frozen mapping.
decimal_string(value: Decimal) -> str
¶
Return the engine's canonical fixed-point decimal form.
decimal_value(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return a finite decimal with at most six fractional places.
execution_quantity(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return an exact signed quantity with six-decimal precision.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
build_initial_state_scenario(*, schemas: TradingEngineContractSchemas, run_id: str, base_currency: str, initial_portfolio: InitialPortfolioState, instruments: Sequence[ExecutionInstrument], venue_calendars: Sequence[Mapping[str, Any]], risk: RiskFinancingRiskPolicy, execution: Mapping[str, Any], max_internal_events: int, metadata: Mapping[str, Any] | None = None, schedule: Sequence[Mapping[str, Any]] = (), slices: Sequence[Mapping[str, Any]] = ()) -> InitialStateScenario
¶
Build and structurally validate one complete scenario v1 document.
initial_state_scenario_to_json(scenario: InitialStateScenario, *, indent: int | None = 2) -> str
¶
Serialize one batch scenario using canonical decimal strings.
initial_state_scenario_to_jsonl(scenario: InitialStateScenario) -> str
¶
Serialize one scenario as bounded-memory v1 JSON Lines records.
write_initial_state_scenario(scenario: InitialStateScenario, path: str | Path, *, stream: bool = False, overwrite: bool = False) -> Path
¶
Write one batch or JSON Lines initial-state scenario without replacement.
reconcile_initial_state_replay(schemas: TradingEngineContractSchemas, scenario_path: str | Path, journal_path: str | Path) -> InitialStateReconciliation
¶
Reconcile opening portfolio and valuation evidence against a v1 scenario.
bind_initial_state_manifest(manifest: Mapping[str, Any], scenario: InitialStateScenario) -> Mapping[str, Any]
¶
Return an immutable replay manifest with explicit v1 opening-state identity.
_validate_portfolio_policy(scenario: InitialStateScenario) -> None
¶
_expected_valuation(scenario: InitialStateScenario) -> dict[str, object]
¶
_stream_records(scenario: InitialStateScenario) -> tuple[dict[str, object], ...]
¶
_scenario_from_document(value: object) -> InitialStateScenario
¶
_portfolio_from_document(value: object) -> InitialPortfolioState
¶
_instrument_from_document(value: object) -> ExecutionInstrument
¶
_risk_from_document(value: object) -> RiskFinancingRiskPolicy
¶
_instrument_payload(value: ExecutionInstrument) -> dict[str, object]
¶
_risk_payload(value: RiskFinancingRiskPolicy) -> dict[str, object]
¶
_require_contract(schemas: TradingEngineContractSchemas) -> None
¶
_mapping(value: object, *, name: str) -> dict[str, object]
¶
_sequence(event: Mapping[str, object]) -> int
¶
_unique(values: Iterable[str], *, name: str) -> None
¶
_bps_ceil(value: Decimal, bps: int) -> Decimal
¶
_canonical_json(value: object) -> bytes
¶
Risk, fees, financing, and settlement¶
persistra.integrations.trading_engine.risk_financing
¶
Typed Trading Engine risk, fees, financing, and settlement contracts.
RISK_FINANCING_CONTRACT_VERSION: Final = '1'
module-attribute
¶
GroupType = Literal['issuer', 'sector', 'currency', 'country', 'asset_class', 'custom']
¶
FeeKind = Literal['fixed', 'notional_bps', 'per_unit']
¶
FeeRounding = Literal['up', 'down', 'nearest']
¶
FeeApplicability = Literal['any', 'maker', 'taker']
¶
DayCount = Literal['actual_365', 'actual_360']
¶
Compounding = Literal['simple', 'daily']
¶
MissingFinancingData = Literal['reject', 'zero']
¶
LocatePolicy = Literal['reject_order', 'clip_fill']
¶
RecallPolicy = Literal['reject_new_shorts', 'close_out']
¶
CashBuyingPower = Literal['total_cash', 'settled_cash']
¶
PositionAvailability = Literal['total_positions', 'settled_positions']
¶
SchemaReplayResult
dataclass
¶
Schema-verified replay envelope and model-specific execution-price evidence.
TradingEngineContractError
¶
Bases: ValueError
A structural contract failure with artifact and schema-path context.
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
EngineCapabilities
dataclass
¶
Machine-readable compatibility surface advertised by an engine executable.
InstrumentRiskPolicy
dataclass
¶
Approved risk limits for exactly one catalog instrument.
RiskGroupLimits
dataclass
¶
Optional v1 exposure ceilings for a named risk group.
RiskGroup
dataclass
¶
Versioned, explicitly sourced group membership and limits.
RiskFinancingRiskPolicy
dataclass
¶
Aggregate and per-instrument v1 risk policy.
FeeComponent
dataclass
¶
One named native-currency fee or rebate component.
InstrumentFeeSchedule
dataclass
¶
Composable fee schedule selected for one instrument.
FeeExecutionPolicy
dataclass
¶
Current completed-bar execution configuration with per-instrument fees.
FinancingPolicy
dataclass
¶
Version-10 financing accrual, locate, and recall choices.
SettlementCalendar
dataclass
¶
Versioned canonical business-date set for settlement.
SettlementRule
dataclass
¶
Business-day settlement lag for one instrument.
SettlementPolicy
dataclass
¶
Buying-power, availability, calendar, and lag choices.
RiskFinancingScenario
dataclass
¶
One immutable, schema-validated Trading Engine v1 scenario.
RiskFinancingReplay
dataclass
¶
Schema replay with reconciled risk, fee, accrual, and settlement evidence.
freeze_portable_mapping(value: Mapping[str, Any], *, name: str, redact_api_keys: bool = False) -> Mapping[str, Any]
¶
Copy and recursively freeze one portable JSON mapping.
thaw_portable_mapping(value: Mapping[str, Any]) -> dict[str, Any]
¶
Return a mutable JSON-compatible copy of one frozen mapping.
decimal_string(value: Decimal) -> str
¶
Return the engine's canonical fixed-point decimal form.
decimal_value(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return a finite decimal with at most six fractional places.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
require_risk_financing_capabilities(capabilities: EngineCapabilities, schemas: TradingEngineContractSchemas, *, scenario_format: Literal['json', 'jsonl'] = 'json') -> None
¶
Reject an engine or schema set that cannot execute the selected v1 contract.
build_risk_financing_scenario(*, schemas: TradingEngineContractSchemas, base_scenario: Mapping[str, Any], risk: RiskFinancingRiskPolicy, execution: FeeExecutionPolicy, financing: FinancingPolicy, settlement: SettlementPolicy) -> RiskFinancingScenario
¶
Select v1 policies, validate alignment, and build batch and stream contracts.
risk_financing_scenario_to_json(scenario: RiskFinancingScenario, *, indent: int | None = 2) -> str
¶
Serialize one v1 batch scenario.
risk_financing_scenario_to_jsonl(scenario: RiskFinancingScenario) -> str
¶
Serialize one v1 scenario as JSON Lines records.
write_risk_financing_scenario(scenario: RiskFinancingScenario, path: str | Path, *, stream: bool = False, overwrite: bool = False) -> Path
¶
Write one selected v1 scenario without replacing artifacts by default.
reconcile_risk_financing_replay(schemas: TradingEngineContractSchemas, scenario_path: str | Path, journal_path: str | Path) -> RiskFinancingReplay
¶
Validate and reconcile every v1 risk, fee, financing, and settlement event.
bind_risk_financing_manifest(manifest: Mapping[str, Any], scenario: RiskFinancingScenario) -> Mapping[str, Any]
¶
Bind exact policy and scenario identities into an immutable replay manifest.
_validate_alignment(document: dict[str, Any], *, risk: RiskFinancingRiskPolicy, execution: FeeExecutionPolicy, settlement: SettlementPolicy) -> None
¶
_validate_slice_observations(document: Mapping[str, Any], *, instrument_ids: set[str], currencies: set[str]) -> None
¶
_stream_records(document: Mapping[str, Any]) -> tuple[dict[str, object], ...]
¶
_reconcile_fill(payload: Mapping[str, object]) -> None
¶
_reconcile_cash_interest(payload: Mapping[str, object]) -> None
¶
_reconcile_borrow_charge(payload: Mapping[str, object]) -> None
¶
_reconcile_settlements(selected: Mapping[str, list[Mapping[str, object]]]) -> None
¶
_reconcile_valuation(payload: Mapping[str, object]) -> None
¶
_require_contract(schemas: TradingEngineContractSchemas) -> None
¶
_mapping(value: object, *, name: str) -> dict[str, object]
¶
_number(value: Mapping[str, object], name: str) -> Decimal
¶
_decimal(value: object) -> str
¶
_basis_points(value: object, *, name: str, positive: bool = False) -> None
¶
_choice(value: object, choices: set[str], *, name: str) -> None
¶
_unique(values: Iterable[object], *, name: str) -> None
¶
_date(value: date | str, *, name: str) -> date
¶
_timestamp(value: object, *, name: str) -> datetime
¶
_canonical_json(value: object) -> bytes
¶
Venue and lifecycle replay¶
persistra.integrations.trading_engine.lifecycle_replay
¶
Explicit Trading Engine venue, corporate-action, and lifecycle replay.
LIFECYCLE_CONTRACT_VERSION: Final = '1'
module-attribute
¶
MAX_LIFECYCLE_EVENTS_PER_SLICE: Final = 1024
module-attribute
¶
SessionPolicy = Literal['regular', 'early_close', 'holiday']
¶
SessionPhase = Literal['premarket', 'opening_auction', 'regular', 'closing_auction', 'postmarket']
¶
DistributionKind = Literal['stock_dividend', 'rights', 'spin_off']
¶
LifecycleKind = Literal['expiration', 'delisting']
¶
LifecycleCorporateAction = SplitLifecycleAction | CashDividendLifecycleAction | DistributionLifecycleAction
¶
LifecycleEvent = HaltLifecycleEvent | ResumeLifecycleEvent | IdentifierChangeLifecycleEvent | TerminalLifecycleEvent
¶
SchemaReplayResult
dataclass
¶
Schema-verified replay envelope and model-specific execution-price evidence.
TradingEngineContractError
¶
Bases: ValueError
A structural contract failure with artifact and schema-path context.
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
LifecycleProvenance
dataclass
¶
Normalized provider identity and adjustment state retained beside an event.
EventDeliveryPolicy
dataclass
¶
Explicit effective, availability, and delivery clocks for one source event.
VenuePhasePolicy
dataclass
¶
One venue phase resolved from local policy to absolute instants.
VenueSessionPolicy
dataclass
¶
Explicit regular, early-close, or holiday policy for one local date.
VenueCalendarPolicy
dataclass
¶
Versioned venue-local policy mapped to explicit UTC session instants.
FractionalEntitlementPolicy
dataclass
¶
Explicit rejection or cash-in-lieu behavior for fractional entitlements.
SplitLifecycleAction
dataclass
¶
CashDividendLifecycleAction
dataclass
¶
DistributionLifecycleAction
dataclass
¶
TerminalDisposition
dataclass
¶
HaltLifecycleEvent
dataclass
¶
ResumeLifecycleEvent
dataclass
¶
IdentifierChangeLifecycleEvent
dataclass
¶
TerminalLifecycleEvent
dataclass
¶
ScheduledCorporateAction
dataclass
¶
ScheduledLifecycleEvent
dataclass
¶
LifecycleSliceEvents
dataclass
¶
LifecycleReplayScenario
dataclass
¶
LifecycleReplayResult
dataclass
¶
freeze_portable_mapping(value: Mapping[str, Any], *, name: str, redact_api_keys: bool = False) -> Mapping[str, Any]
¶
Copy and recursively freeze one portable JSON mapping.
thaw_portable_mapping(value: Mapping[str, Any]) -> dict[str, Any]
¶
Return a mutable JSON-compatible copy of one frozen mapping.
decimal_string(value: Decimal) -> str
¶
Return the engine's canonical fixed-point decimal form.
decimal_value(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return a finite decimal with at most six fractional places.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
require_lifecycle_capabilities(capabilities: Mapping[str, Any], schemas: TradingEngineContractSchemas, *, scenario_format: Literal['json', 'jsonl'] = 'json') -> None
¶
Require the first engine contract carrying complete lifecycle semantics.
build_lifecycle_replay_scenario(*, schemas: TradingEngineContractSchemas, base_scenario: Mapping[str, Any], calendars: Sequence[VenueCalendarPolicy], slices: Sequence[LifecycleSliceEvents]) -> LifecycleReplayScenario
¶
Build a schema-validated v1 scenario from deliberate executable policies.
lifecycle_scenario_to_json(scenario: LifecycleReplayScenario, *, indent: int | None = 2) -> str
¶
lifecycle_scenario_to_jsonl(scenario: LifecycleReplayScenario) -> str
¶
write_lifecycle_scenario(scenario: LifecycleReplayScenario, path: str | Path, *, stream: bool = False, overwrite: bool = False) -> Path
¶
bind_lifecycle_manifest(manifest: Mapping[str, Any], scenario: LifecycleReplayScenario) -> Mapping[str, Any]
¶
reconcile_lifecycle_replay(schemas: TradingEngineContractSchemas, scenario_path: str | Path, journal_path: str | Path) -> LifecycleReplayResult
¶
Reconcile declared actions and lifecycle transitions with their accounting evidence.
_validate_slice_semantics(data: LifecycleSliceEvents, *, instruments: Mapping[str, str], states: dict[str, str]) -> None
¶
_validate_delivery_bounds(market_slice: Mapping[str, object], data: LifecycleSliceEvents) -> None
¶
_validate_calendar_coverage(calendars: Sequence[VenueCalendarPolicy], *, instruments: set[str]) -> None
¶
_declared_events(document: Mapping[str, object]) -> tuple[dict[str, Mapping[str, object]], dict[str, Mapping[str, object]], dict[str, Mapping[str, object]]]
¶
_reconcile_received(payload: Mapping[str, object], expected: dict[str, Mapping[str, object]]) -> None
¶
_reconcile_action(payload: Mapping[str, object], declared: dict[str, Mapping[str, object]], *, event_type: str) -> None
¶
_reconcile_lifecycle(payload: Mapping[str, object], declared: dict[str, Mapping[str, object]]) -> None
¶
_stream_records(document: Mapping[str, Any]) -> tuple[dict[str, object], ...]
¶
_instruments(document: Mapping[str, Any]) -> dict[str, str]
¶
_source_policy(delivery: EventDeliveryPolicy, provenance: LifecycleProvenance) -> None
¶
_scheduled_dict(event_id: str, delivery: EventDeliveryPolicy, provenance: LifecycleProvenance) -> dict[str, object]
¶
_action_common(value: SplitLifecycleAction | CashDividendLifecycleAction | DistributionLifecycleAction) -> None
¶
_event_common(value: HaltLifecycleEvent | ResumeLifecycleEvent | IdentifierChangeLifecycleEvent | TerminalLifecycleEvent) -> None
¶
_reason(value: object) -> str
¶
_require_contract(schemas: TradingEngineContractSchemas) -> None
¶
_mapping(value: object, *, name: str) -> dict[str, object]
¶
_portable(value: object) -> object
¶
_strings(value: object, *, name: str) -> tuple[str, ...]
¶
_timestamp(value: object, *, name: str) -> datetime
¶
_timestamp_string(value: datetime) -> str
¶
_date(value: object, *, name: str) -> date
¶
_number(value: Mapping[str, object], name: str) -> Decimal
¶
_decimal(value: object) -> str
¶
_basis_points(value: object, *, name: str) -> None
¶
_choice(value: object, choices: set[str], *, name: str) -> None
¶
_unique(values: Iterable[object], *, name: str) -> None
¶
_canonical_json(value: object) -> bytes
¶
Market-data replay¶
persistra.integrations.trading_engine.market_data_replay
¶
Causal Trading Engine quote, trade, and order-book replay contracts.
MARKET_DATA_CONTRACT_VERSION: Final = '1'
module-attribute
¶
MAX_MARKET_EVENTS_PER_SLICE: Final = 4096
module-attribute
¶
MarketDataModel = Literal['quote_trade_v1', 'order_book_v1']
¶
AggressorSide = Literal['buy', 'sell', 'unknown']
¶
BookSide = Literal['bid', 'ask']
¶
QuoteTradeObservation = ExecutableQuote | ExecutableTrade
¶
OrderBookObservation = OrderBookSnapshot | OrderBookSet | OrderBookDelete | OrderBookTrade
¶
SchemaReplayResult
dataclass
¶
Schema-verified replay envelope and model-specific execution-price evidence.
TradingEngineContractError
¶
Bases: ValueError
A structural contract failure with artifact and schema-path context.
TradingEngineContractSchemas
dataclass
¶
One immutable, fingerprinted Trading Engine contract schema set.
load(directory: str | Path) -> TradingEngineContractSchemas
classmethod
¶
Load and verify the authoritative schemas in one version directory.
validate_scenario(value: object) -> None
¶
Validate one batch scenario structurally.
validate_stream_record(value: object, *, line_number: int) -> None
¶
Validate one scenario-stream record structurally.
validate_journal_record(value: object, *, line_number: int) -> None
¶
Validate one journal record structurally.
validate_journal(path: str | Path) -> int
¶
Stream and validate a journal, returning its record count.
read_replay(scenario_path: str | Path, journal_path: str | Path) -> SchemaReplayResult
¶
Validate and reconcile a schema-versioned replay without older semantic adapters.
InstrumentFeeSchedule
dataclass
¶
Composable fee schedule selected for one instrument.
ObservationProvenance
dataclass
¶
Provider and normalized-dataset ordering retained beside an engine event.
ReplayEventClock
dataclass
¶
Economic, availability, receipt, and ingest ordering for one observation.
ExecutableQuote
dataclass
¶
Two-sided displayed quote with positive executable liquidity.
ExecutableTrade
dataclass
¶
Print with explicit aggressor classification and executable quantity.
OrderBookLevel
dataclass
¶
One absolute level-two price and displayed quantity.
OrderBookSnapshot
dataclass
¶
Complete per-instrument opening book for one slice.
OrderBookSet
dataclass
¶
Absolute set update for one side and price.
OrderBookDelete
dataclass
¶
Delete update for an explicitly existing side and price.
OrderBookTrade
dataclass
¶
Aggressor-classified depth-consuming book trade.
ReplaySliceMarketData
dataclass
¶
Causally ordered executable observations assigned to one market slice.
MarketDataExecutionPolicy
dataclass
¶
Selected v1 model, fee schedules, and bounded depth configuration.
MarketDataModelCapability
dataclass
¶
Required configuration and normalized data declared for one engine model.
MarketDataReplayScenario
dataclass
¶
Immutable schema-v1 scenario plus lossless observation provenance.
MarketDataReplayResult
dataclass
¶
Schema replay with source-matched model-specific fills.
freeze_portable_mapping(value: Mapping[str, Any], *, name: str, redact_api_keys: bool = False) -> Mapping[str, Any]
¶
Copy and recursively freeze one portable JSON mapping.
thaw_portable_mapping(value: Mapping[str, Any]) -> dict[str, Any]
¶
Return a mutable JSON-compatible copy of one frozen mapping.
decimal_string(value: Decimal) -> str
¶
Return the engine's canonical fixed-point decimal form.
decimal_value(value: object, *, name: str, positive: bool = False, nonnegative: bool = False) -> Decimal
¶
Return a finite decimal with at most six fractional places.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
market_data_model_capabilities(value: Mapping[str, Any]) -> tuple[MarketDataModelCapability, ...]
¶
Parse the exact market-data execution capability declarations.
require_market_data_capabilities(capabilities: Mapping[str, Any], schemas: TradingEngineContractSchemas, policy: MarketDataExecutionPolicy, *, scenario_format: Literal['json', 'jsonl'] = 'json') -> None
¶
Negotiate model, configuration, contract, format, and data requirements.
build_market_data_replay_scenario(*, schemas: TradingEngineContractSchemas, base_scenario: Mapping[str, Any], execution: MarketDataExecutionPolicy, slices: Sequence[ReplaySliceMarketData]) -> MarketDataReplayScenario
¶
Map executable observations into a validated bounded v1 scenario and stream.
market_data_scenario_to_json(scenario: MarketDataReplayScenario, *, indent: int | None = 2) -> str
¶
Serialize one v1 batch market-data scenario.
market_data_scenario_to_jsonl(scenario: MarketDataReplayScenario) -> str
¶
Serialize one bounded v1 scenario stream.
write_market_data_scenario(scenario: MarketDataReplayScenario, path: str | Path, *, stream: bool = False, overwrite: bool = False) -> Path
¶
Write a batch or streaming scenario without replacement by default.
bind_market_data_manifest(manifest: Mapping[str, Any], scenario: MarketDataReplayScenario) -> Mapping[str, Any]
¶
Bind selected model and lossless normalized observation provenance.
reconcile_market_data_replay(schemas: TradingEngineContractSchemas, scenario_path: str | Path, journal_path: str | Path) -> MarketDataReplayResult
¶
Reconcile received observations and every fill to executable source liquidity.
_validate_observations(data: ReplaySliceMarketData, *, instruments: Mapping[str, tuple[Decimal, Decimal]], execution: MarketDataExecutionPolicy) -> None
¶
_validate_book_updates(events: Sequence[OrderBookObservation], *, execution: MarketDataExecutionPolicy) -> None
¶
_validate_observation_order(observations: Sequence[QuoteTradeObservation | OrderBookObservation]) -> None
¶
_validate_slice_bounds(item: Mapping[str, object], data: ReplaySliceMarketData) -> None
¶
_validate_model_families(slices: Sequence[ReplaySliceMarketData], *, model: MarketDataModel) -> None
¶
_reconcile_received_slice(payload: Mapping[str, object], expected: dict[str, Mapping[str, object]]) -> None
¶
_match_fill_source(fill: Mapping[str, object], slices: Sequence[Mapping[str, object]], *, model: MarketDataModel, remaining_liquidity: dict[tuple[object, ...], Decimal]) -> Mapping[str, object]
¶
_normalized_event(value: Mapping[str, object]) -> dict[str, object]
¶
_instrument_contracts(document: Mapping[str, Any]) -> dict[str, tuple[Decimal, Decimal]]
¶
_stream_records(document: Mapping[str, Any]) -> tuple[dict[str, object], ...]
¶
_book_common(value: OrderBookSnapshot | OrderBookSet | OrderBookDelete | OrderBookTrade) -> None
¶
_require_ingested_after_receipt(clock: ReplayEventClock, provenance: ObservationProvenance) -> None
¶
_require_contract(schemas: TradingEngineContractSchemas) -> None
¶
_mapping(value: object, *, name: str) -> dict[str, object]
¶
_strings(value: object, *, name: str) -> tuple[str, ...]
¶
_timestamp(value: object, *, name: str) -> datetime
¶
_timestamp_string(value: datetime) -> str
¶
_number(value: Mapping[str, object], name: str) -> Decimal
¶
_decimal(value: object) -> str
¶
_basis_points(value: object, *, name: str) -> None
¶
_choice(value: object, choices: set[str], *, name: str) -> None
¶
_unique(values: Iterable[object], *, name: str) -> None
¶
_canonical_json(value: object) -> bytes
¶
Diagnostics and structured results¶
persistra.integrations.trading_engine.diagnostics
¶
Decode versioned Trading Engine diagnostics without parsing prose.
__all__ = ['trading_engine_diagnostic_from_json']
module-attribute
¶
_CONTEXT_FIELDS = {'json_path', 'line', 'sequence', 'event_id', 'order_id', 'causation_ids'}
module-attribute
¶
_CAUSE_FIELDS = {'kind', 'message', 'operation', 'target'}
module-attribute
¶
TradingEngineDiagnostic
dataclass
¶
Version-1 machine-readable failure reported by Trading Engine.
TradingEngineDiagnosticCause
dataclass
¶
Sanitized exception cause supplied by Trading Engine.
TradingEngineDiagnosticContext
dataclass
¶
Typed location and causality fields from one engine diagnostic.
exact_fields(value: object, expected: set[str], *, name: str) -> dict[str, Any]
¶
Validate and narrow an exact JSON object.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
trading_engine_diagnostic_from_json(document: str) -> TradingEngineDiagnostic
¶
Decode one strict version-1 Trading Engine diagnostic document.
diagnostic_from_object(value: object) -> TradingEngineDiagnostic
¶
_context_from_object(value: object) -> TradingEngineDiagnosticContext
¶
_cause_from_object(value: object) -> TradingEngineDiagnosticCause
¶
_optional_fields(value: object, allowed: set[str], *, name: str, allow_additive: bool = False) -> dict[str, Any]
¶
_string(value: object, *, name: str, nonempty: bool = False) -> str
¶
_optional_string(value: object, *, name: str) -> str | None
¶
_optional_quantity(value: object, *, name: str) -> int | None
¶
_unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]
¶
persistra.integrations.trading_engine.automation
¶
Structured Trading Engine CLI success and failure artifacts.
TRADING_ENGINE_RESULT_VERSION: Final = '1'
module-attribute
¶
_HASH_FIELDS = {'scenario_sha256', 'journal_sha256', 'strategy_transcript_sha256'}
module-attribute
¶
_COUNT_FIELDS = {'instruments', 'schedule_batches', 'slices', 'audits', 'orders', 'active_orders', 'filled_orders', 'rejected_orders'}
module-attribute
¶
_TERMINAL_VALUATION_FIELDS = {'base_currency', 'cash', 'net_market_value', 'long_market_value', 'short_market_value', 'gross_exposure', 'cost_basis', 'realized_pnl', 'unrealized_pnl', 'equity', 'dividend_pnl', 'execution_fees', 'borrow_fees', 'total_fees'}
module-attribute
¶
StrategyResponseRejection
dataclass
¶
Version-1 diagnostic record ending one failed strategy transcript.
TradingEngineDiagnostic
dataclass
¶
Version-1 machine-readable failure reported by Trading Engine.
TradingEngineProcessError
dataclass
¶
Bases: RuntimeError
A validation or replay subprocess failed.
TradingEngineSuccessHashes
dataclass
¶
Content identities reported by one successful CLI operation.
TradingEngineSuccessCounts
dataclass
¶
Bounded nonnegative replay counts reported by Trading Engine.
TradingEngineSuccessSummary
dataclass
¶
Versioned machine-readable validation or replay success.
StructuredEngineFailureStatus
dataclass
¶
Safe manifest-ready status derived from a typed process failure.
freeze_portable_mapping(value: Mapping[str, Any], *, name: str, redact_api_keys: bool = False) -> Mapping[str, Any]
¶
Copy and recursively freeze one portable JSON mapping.
thaw_portable_mapping(value: Mapping[str, Any]) -> dict[str, Any]
¶
Return a mutable JSON-compatible copy of one frozen mapping.
exact_fields(value: object, expected: set[str], *, name: str) -> dict[str, Any]
¶
Validate and narrow an exact JSON object.
identifier(value: object, *, name: str) -> str
¶
Validate an engine identifier or nonempty label.
quantity_value(value: object, *, name: str, positive: bool = False) -> int
¶
Return a checked nonnegative whole int64 quantity.
trading_engine_success_from_json(document: str) -> TradingEngineSuccessSummary
¶
Parse a strict CLI result v1 success document without matching prose.
verify_trading_engine_success(summary: TradingEngineSuccessSummary, scenario_path: str | Path, *, journal_path: str | Path | None = None, strategy_transcript_path: str | Path | None = None) -> None
¶
Cross-check one success summary against immutable scenario and output artifacts.
structured_engine_failure(error: object) -> StructuredEngineFailureStatus
¶
Convert a process exception into bounded status without retaining stdout or stderr.
bind_engine_status_manifest(manifest: Mapping[str, Any], status: TradingEngineSuccessSummary | StructuredEngineFailureStatus) -> Mapping[str, Any]
¶
Attach one structured success or failure status to a replay manifest.