Vector¶

Exported from sekai and sekai.vector (shared from sekai.vector only). See the Vector environments guide.

Vector environments¶

VecEnv ¶

Bases: ABC, Generic[ObsType, ActType]

Abstract base for vectorised environments.

Runs N environments in parallel and returns batched results.

Key differences from gymnasium's VectorEnv: - observation_space is always the single-env space - batch_observation_space is always the batched space - step/reset return structured dataclasses, not tuples - infos is list[dict], not dict[str, ndarray]

Autoreset (on by default): when a sub-env finishes an episode it is reset in the same step() call. The returned observation for that env is the first observation of the new episode; the terminal observation and info are in infos[i]["final_observation"] and infos[i]["final_info"], and the reset info in infos[i]["reset_info"]. This matches the AutoReset wrapper.

Without autoreset (autoreset=False), a sub-env that finishes an episode is frozen until the next reset(): it is not stepped again and its action is ignored, and every later step() returns, for it, its final observation, reward 0, its final terminated / truncated flags (so dones stays True) and an empty info. A sub-env that resets itself (e.g. wrapped in AutoReset, so its info carries final_observation) is not frozen.

Every VecEnv has async_step()/async_reset(); by default they run the sync methods (AsyncVecEnv overrides them to run sub-envs concurrently). As with Env, a subclass that overrides only step()/reset() gets async methods that run its overrides.

batch_observation_space `property` ¶

batch_observation_space: Space[Any]

The space of a batched observation, with a leading num_envs axis.

batch_action_space `property` ¶

batch_action_space: Space[Any]

The space of a batched action, with a leading num_envs axis.

step `abstractmethod` ¶

step(actions: Any) -> VecStepResult[Any]

Step all environments with a batch of actions.

reset `abstractmethod` ¶

reset(*, seed: list[SeedType] | SeedType = None, options: list[dict[str, Any] | None] | dict[str, Any] | None = None) -> VecResetResult[Any]

Reset all environments.

seed: an int derives one independent seed per sub-env; a list gives each sub-env its own seed; None continues each sub-env's RNG stream. options: one dict for all sub-envs, or a list with one per sub-env.

close `abstractmethod` ¶

close() -> None

Close all sub-environments and release resources.

Every sub-env is closed even if closing another one raises; the first such error is re-raised once everything is released.

async_step `async` ¶

async_step(actions: Any) -> VecStepResult[Any]

Async step; the default runs step().

async_reset `async` ¶

async_reset(*, seed: list[SeedType] | SeedType = None, options: list[dict[str, Any] | None] | dict[str, Any] | None = None) -> VecResetResult[Any]

Async reset; the default runs reset().

call ¶

call(method: str, *args: Any, **kwargs: Any) -> list[Any]

Call a method (or read an attribute) on each sub-environment.

Like get_attr(), the name is looked up through each sub-env's wrapper stack.

get_attr ¶

get_attr(name: str) -> list[Any]

Get an attribute from each sub-environment, searching its wrapper stack.

set_attr ¶

set_attr(name: str, values: list[Any] | Any) -> None

Set an attribute on each sub-environment (one value, or a list of num_envs).

The value goes to the outermost layer of each sub-env's wrapper stack that has the attribute, as with Wrapper.set_attr().

SyncVecEnv ¶

SyncVecEnv(env_fns: Sequence[Callable[[], Env[ObsType, ActType]]], *, autoreset: bool = True, copy: bool = True)

Bases: VecEnv[ObsType, ActType]

Vectorised env that runs N sub-environments sequentially in a single process.

Best for fast environments where multiprocessing overhead would dominate. Use ProcVecEnv for CPU-bound simulators.

Parameters:

Name	Type	Description	Default
`env_fns`	`Sequence[Callable[[], Env[ObsType, ActType]]]`	Callables that each build one sub-environment. If one raises, the sub-envs already built are closed before the error propagates.	required
`autoreset`	`bool`	Reset sub-envs automatically when their episode ends (see VecEnv).	`True`
`copy`	`bool`	Kept for backwards compatibility. Batched observations are always freshly allocated arrays, so this has no effect.	`True`

AsyncVecEnv ¶

AsyncVecEnv(env_fns: Sequence[Callable[[], Env[ObsType, ActType]]], *, autoreset: bool = True)

Bases: VecEnv[ObsType, ActType]

Vectorised env using asyncio to run N environments concurrently.

Unlike ProcVecEnv (multiprocessing), this uses asyncio.gather — ideal for I/O-bound environments such as live market data feeds, REST-based simulators, or any env with async_step/async_reset overrides.

CPU-bound envs with blocking step() calls will NOT benefit from this; use ProcVecEnv instead.

From async code, use await vec.async_step(...) / await vec.async_reset(...). The sync step()/reset() run those coroutines on a private event loop owned by this VecEnv, so they must not be called from inside a running loop.

If an env_fn raises, the sub-envs already built are closed before the error propagates.

ProcVecEnv ¶

ProcVecEnv(env_fns: Sequence[Callable[[], Env[ObsType, ActType]]], *, autoreset: bool = True, context: str | None = None, close_timeout: float = 5.0, num_workers: int | None = None, shared_memory: bool | None = None)

Bases: VecEnv[ObsType, ActType]

Vectorised env using worker processes for true CPU-bound parallelism.

Sub-environments run in worker processes and communicate via multiprocessing Pipes. Use this for CPU-intensive simulators; for I/O-bound or async environments use AsyncVecEnv, and for simulators that vectorise with array operations prefer a native vector env (see sekai.make_vec).

Two things keep the inter-process overhead low:

Shared memory (on by default when the observation space has fixed shapes: Box, Discrete, MultiDiscrete, MultiBinary and Dict/Tuple/MaskedBox of them). Workers write observations straight into one shared block, so only rewards, flags and infos are pickled. Large observations (order books, multi-asset windows) no longer pay a serialisation cost per step.
Several sub-envs per worker (num_workers): with more sub-envs than CPU cores, grouping them avoids oversubscription and amortises each round trip over several envs.

The default start method is "forkserver" ("spawn" on Windows), so the script that creates a ProcVecEnv must guard its entry point with if __name__ == "__main__":. env_fns are sent with cloudpickle, so lambdas and locally defined classes work.

An exception inside a sub-env is re-raised in the parent as VecEnvWorkerError, carrying the sub-env index and the worker's traceback. The other workers' replies are still collected, so the VecEnv stays usable afterwards (the step results of that call are discarded, including those of sub-envs that did advance); replies left unread by an interrupted call (e.g. Ctrl-C, which workers ignore and leave to the parent) are discarded before the next command. A worker that dies also raises VecEnvWorkerError: its rank is the sub-env it ran, or, for a worker that ran several (so which one failed is unknown), the first of them, and the message names the worker and all its sub-envs.

Parameters:

Name	Type	Description	Default
`env_fns`	`Sequence[Callable[[], Env[ObsType, ActType]]]`	Callables that each build one sub-environment (in its worker). If one raises (or anything else in the constructor fails), the sub-envs already built are closed and the workers stopped before the error propagates.	required
`autoreset`	`bool`	Reset sub-envs automatically when their episode ends (see VecEnv).	`True`
`context`	`str \| None`	multiprocessing start method; defaults to "forkserver" ("spawn" on Windows).	`None`
`close_timeout`	`float`	Seconds close() waits for workers before terminating them.	`5.0`
`num_workers`	`int \| None`	Worker processes; sub-envs are split evenly between them. Defaults to one process per sub-env.	`None`
`shared_memory`	`bool \| None`	Pass observations through shared memory. None (default) enables it when the observation space allows; True requires it.	`None`

shared_memory `property` ¶

shared_memory: bool

Whether observations travel through shared memory.

close ¶

close() -> None

Close every sub-env, waiting up to close_timeout seconds before terminating workers.

Workers close all their sub-envs even if one fails; the first failure is re-raised as VecEnvWorkerError once every worker has stopped.

NativeVecEnv ¶

NativeVecEnv(num_envs: int, observation_space: Space[ObsType], action_space: Space[ActType], *, autoreset: bool = True, max_episode_steps: int | None = None)

Bases: VecEnv[ObsType, ActType]

Base class for environments that simulate every sub-env at once with array operations.

SyncVecEnv steps N separate Python environments one after another; a native vector env keeps the state of all N in arrays with a leading num_envs axis and advances them together, which is typically 10-100x faster for simulators that vectorise naturally (most market models do). Register one with sekai.register(..., vector_entry_point=...) and sekai.make_vec() uses it automatically.

Subclasses implement:

_reset_envs(indices, options): (re)initialise the sub-envs in indices (an int array), with one options dict (or None) per index.
_observe(): the batched observation of all sub-envs.
_step(actions): advance every sub-env; return (rewards, terminated, truncated, infos) where the first three have shape (num_envs,) and infos maps keys to arrays with a leading num_envs axis.
optionally _reset_infos(indices): info arrays for freshly reset sub-envs.

The base class provides seeding (one numpy.random.Generator, self.rng, shared by all sub-envs), per-env options, same-step autoreset with final_observation / final_info / reset_info in the infos (the contract of every sekai VecEnv), conversion of info arrays to per-env dicts, ResetRequired from a step() before the first reset(), and an optional time limit: with max_episode_steps=n every sub-env is truncated after n steps and its infos carry elapsed_steps, exactly as TimeLimit does for each sub-env of a SyncVecEnv. The elapsed_steps property counts each sub-env's steps, limit or not.

With autoreset=False, a sub-env that finishes is frozen until the next reset(), as in every VecEnv: step() returns its final observation, reward 0, its final terminated / truncated flags and an empty info, whatever _step() computes for it. _step() still advances every sub-env, so it is passed a frozen sub-env's last action again (when actions are a numeric array), and self.frozen tells which sub-envs are frozen: a simulation that would read past its data, or that should keep a finished sub-env's state (e.g. for get_attr), must leave those sub-envs as they are.

Because all sub-envs share one generator, a native env is not draw-for-draw identical to the same env under SyncVecEnv; it is identical in distribution, and exactly identical whenever the dynamics are deterministic given the options (e.g. historical data with fixed starts). reset(seed=...) takes what every VecEnv takes: an int seeds the generator; a list of num_envs seeds is combined into one seed, where a None entry contributes a draw from the current stream (so a list of only None continues it, like seed=None); a Generator raises TypeError.

get_attr(name) / call(name) read attributes and methods of this object: an array result with a leading num_envs axis is split per env, anything else is repeated for every env, and so is everything a method or property marked with @shared returns. set_attr(name, values) changes an existing attribute (a list of num_envs values is stored as one array) and raises AttributeError for a name this object does not have, since a new attribute would have no effect on the simulation.

rng `property` ¶

rng: Generator

The generator shared by all sub-envs (created from OS entropy if unseeded).

_reset_envs `abstractmethod` ¶

_reset_envs(indices: NDArray[intp], options: Sequence[dict[str, Any] | None]) -> None

_observe `abstractmethod` ¶

_observe() -> Any

_step `abstractmethod` ¶

_step(actions: Any) -> tuple[NDArray[Any], NDArray[Any], NDArray[Any], InfoArrays]

_reset_infos ¶

_reset_infos(indices: NDArray[intp]) -> InfoArrays

shared ¶

shared(member: _F) -> _F

Mark a method (or property getter) of a NativeVecEnv as shared by every sub-env.

get_attr() / call() then give its value to every sub-env as it is, instead of splitting an array with a leading num_envs axis, which would mistake e.g. a schedule of n_steps rows for per-env values whenever n_steps == num_envs.

Example

class MyVecEnv(NativeVecEnv):
    @shared
    def schedule(self) -> np.ndarray:  # (n_steps, 1), the same for every sub-env
        ...

    @property
    @shared
    def tick_sizes(self) -> np.ndarray: ...

Results¶

VecStepResult `dataclass` ¶

VecStepResult(observations: ObsType, rewards: NDArray[float64], terminated: NDArray[bool_], truncated: NDArray[bool_], infos: list[InfoDict])

Bases: Generic[ObsType]

Structured result from VecEnv.step().

observations is the batch obs with shape (N, *obs_shape). infos is a list of per-env InfoDicts (not stacked numpy arrays).

VecResetResult `dataclass` ¶

VecResetResult(observations: ObsType, infos: list[InfoDict])

Bases: Generic[ObsType]

Structured result from VecEnv.reset(). dones is all False, mirroring VecStepResult.

stack_infos ¶

stack_infos(infos: list[InfoDict]) -> dict[str, Any]

Stack a list of per-env info dicts into {key: np.array}.

Keys that appear in all dicts are stacked; missing keys are skipped. Useful when a downstream algorithm wants numpy-array info values.

Wrapper base classes¶

VecWrapper ¶

VecWrapper(venv: VecEnv[ObsType, ActType])

Bases: VecEnv[WObsType, WActType], Generic[WObsType, WActType, ObsType, ActType]

Transparent wrapper around a VecEnv.

Like Wrapper, observation_space / action_space delegate to the wrapped VecEnv unless assigned (always the single-env spaces). Both the sync and async APIs are delegated, so wrapping an AsyncVecEnv keeps await vec.async_step(...) concurrent.

VecObservationWrapper ¶

VecObservationWrapper(venv: VecEnv[ObsType, ActType])

Bases: VecWrapper[WObsType, ActType, ObsType, ActType], Generic[WObsType, ActType, ObsType]

Applies a transformation to every batch of observations.

Terminal observations that autoreset stores in infos[i]["final_observation"] are transformed too, via final_observation(), so value bootstrapping sees the same representation as the policy.

final_observation ¶

final_observation(obs: Any) -> Any

Transform one sub-env's terminal observation (from autoreset infos).

The default runs observation() on a batch of one. Override it if your transform is not valid for a single-row batch or has side effects (e.g. updating running statistics) that terminal observations should not cause.

VecRewardWrapper ¶

VecRewardWrapper(venv: VecEnv[ObsType, ActType])

Bases: VecWrapper[ObsType, ActType, ObsType, ActType], Generic[ObsType, ActType]

Applies a transformation to every batch of rewards.

VecActionWrapper ¶

VecActionWrapper(venv: VecEnv[ObsType, ActType])

Bases: VecWrapper[ObsType, WActType, ObsType, ActType], Generic[ObsType, WActType, ActType]

Applies a transformation to every batch of actions.

Ready-made wrappers¶

VecRecordEpisodeStatistics ¶

VecRecordEpisodeStatistics(venv: VecEnv[ObsType, ActType], tracker: StatsTracker | None = None, deque_size: int = 100)

Bases: VecWrapper[ObsType, ActType, ObsType, ActType]

Per-sub-env episode statistics for a VecEnv.

When sub-env i finishes an episode, infos[i]["episode"] is an EpisodeStats, exactly as with the single-env RecordEpisodeStatistics, and the stats are reported to the tracker. Works with autoreset on or off: without autoreset, a finished sub-env is frozen (see VecEnv) and keeps reporting done, so it is ignored until the next reset() and stepping the others on (e.g. while not finished.all()) records each episode once.

VecNormalizeObservation ¶

VecNormalizeObservation(venv: VecEnv[Any, ActType], epsilon: float = 1e-08, *, stats: StatsKind = 'running', halflife: float = 1000.0, clip: float | None = None)

Bases: VecObservationWrapper[Any, ActType, Any]

Normalises observations with running statistics shared across all sub-envs.

The vectorised counterpart of NormalizeObservation, with the same options: any observation space (Box leaves normalised, MaskedBox leaves over occupied slots only, other leaves passed through), stats="running" or "ewm" (the halflife counts vector steps), and optional clipping. infos[i]["final_observation"] from autoreset is normalised too, with the current statistics but without updating them. Set update_running_mean = False to freeze the statistics.

obs_rms `property` ¶

obs_rms: Any

The running statistics (nested like the observation space).

state_dict ¶

state_dict() -> Any

The normalisation statistics as plain data.

load_state_dict ¶

load_state_dict(state: Any) -> None

Restore statistics saved by state_dict().

VecNormalizeReward ¶

VecNormalizeReward(venv: VecEnv[ObsType, ActType], gamma: float = 0.99, epsilon: float = 1e-08)

Bases: VecWrapper[ObsType, ActType, ObsType, ActType]

Scales rewards by the running std of each sub-env's discounted return.

The vectorised counterpart of NormalizeReward (OpenAI baselines' VecNormalize reward scaling). Returns are reset per sub-env when its episode ends. Set update_running_mean = False to freeze the statistics.

VecTransformObservation ¶

VecTransformObservation(venv: VecEnv[Any, ActType], func: Callable[[Any], Any], new_observation_space: Space[Any] | None = None)

Bases: VecObservationWrapper[Any, ActType, Any]

Applies func to every batch of observations.

Example

vec = VecTransformObservation(vec, lambda obs: obs / 255.0)

VecTransformReward ¶

VecTransformReward(venv: VecEnv[ObsType, ActType], func: Callable[[NDArray[float64]], NDArray[float64]])

Bases: VecRewardWrapper[ObsType, ActType]

Applies func to every batch of rewards (an (N,) float64 array).

Example

vec = VecTransformReward(vec, np.sign)

Vector¶

Vector environments¶

VecEnv ¶

batch_observation_space property ¶

batch_action_space property ¶

step abstractmethod ¶

reset abstractmethod ¶

close abstractmethod ¶

async_step async ¶

async_reset async ¶

call ¶

get_attr ¶

set_attr ¶

SyncVecEnv ¶

AsyncVecEnv ¶

ProcVecEnv ¶

shared_memory property ¶

close ¶

NativeVecEnv ¶

rng property ¶

_reset_envs abstractmethod ¶

_observe abstractmethod ¶

_step abstractmethod ¶

_reset_infos ¶

shared ¶

Results¶

VecStepResult dataclass ¶

VecResetResult dataclass ¶

stack_infos ¶

Wrapper base classes¶

VecWrapper ¶

VecObservationWrapper ¶

final_observation ¶

VecRewardWrapper ¶

VecActionWrapper ¶

Ready-made wrappers¶

VecRecordEpisodeStatistics ¶

VecNormalizeObservation ¶

obs_rms property ¶

state_dict ¶

load_state_dict ¶

VecNormalizeReward ¶

VecTransformObservation ¶

VecTransformReward ¶

batch_observation_space `property` ¶

batch_action_space `property` ¶

step `abstractmethod` ¶

reset `abstractmethod` ¶

close `abstractmethod` ¶

async_step `async` ¶

async_reset `async` ¶

shared_memory `property` ¶

rng `property` ¶

_reset_envs `abstractmethod` ¶

_observe `abstractmethod` ¶

_step `abstractmethod` ¶

VecStepResult `dataclass` ¶

VecResetResult `dataclass` ¶

obs_rms `property` ¶