PyActuator

The actuation core of the Optophi family. PyActuator is a stateless, multi-domain Python package for performing side-effecting operations on external systems on behalf of strategy, state-machine, and agent components. Trading is one domain among several — Google Workspace, Comms (Slack / Email), Documents (PDF / Markdown / HTML), and Database (Postgres, SQLite, MongoDB, Redis) are first-class siblings.

Features

• Multi-domain — 15 domains under pyactuator.domains.*, each with the same shape (types, protocols, errors, adapters/): market data (alpaca, fmp, massive, polymarket, finance), execution (trading), workspace/comms (google, comms, documents), data (database, datasets, data_sources, analytics, charts), and research. See Domains Overview.
• Stateless core — pyactuator.core ships envelopes, idempotency, retry, audit, policy, registry, and spec primitives. No I/O.
• Adapter pattern — every external system is an adapter behind a Protocol. Mock adapters ship with every domain.
• Bridges — install pyactuator operations into PyStator state machines (integrations.pystator) and PyGubernator tool registries (integrations.pygubernator) with one call.
• CLI — pyactuator run for unified YAML/JSON configs, pyactuator tools list for valid operation names, plus api, ui, worker, and docs helpers.
• HTTP API — domain-scoped FastAPI routers under /api/v1/<domain>/..., sharing a single AdapterRegistry.
• Optional deps — alpaca-py, httpx, google-auth, jinja2, markdown-it-py, and reportlab are all optional extras, so you install only the vendors you actuate. The base install pulls duckdb + pyarrow for the analytics, datasets, data-source, and charts domains.
• Backwards compatible — existing from pyactuator import OrderRequest imports continue to work as re-exports of the trading domain.

Installation

# Core only (envelopes, registry, mock adapters) — no mandatory dependencies
pip install pyactuator

# Data engine (analytics / datasets / data sources / charts) — DuckDB + PyArrow
pip install pyactuator[analytics]

# Trading with Alpaca broker
pip install pyactuator[trading-alpaca]

# Google Workspace (Drive / Sheets / Docs)
pip install pyactuator[google]

# Slack via webhook or bot token
pip install pyactuator[comms-slack]

# PDF + Markdown rendering
pip install pyactuator[documents-pdf,documents-md]

# Local file read (PDF text extraction + images; pypdf + Pillow)
pip install pyactuator[documents-read]

# Everything
pip install pyactuator[all]

# Development
pip install -e ".[dev]"

Documentation

Published docs (Material for MkDocs) live at optophi.github.io/pyactuator.

To build or preview locally:

pip install pyactuator[docs]   # or: pip install -e ".[dev]" (includes docs tooling)
mkdocs serve

The site configuration mirrors PyStator / PyCharter: mkdocs.yml, mkdocs.optophi.yml for the Optophi.com docs deployment, and docs/ as the documentation root. CI runs mkdocs build --strict when mkdocs.yml is present (see ./scripts/ci.sh).

Quick start: invoke (the config-driven entry point)

pyactuator.sync.invoke / pyactuator.invoke dispatch any catalogued operation by name against a mock (or real) AdapterRegistry and return the canonical wire envelope — the recommended entry point when you don't need a specific domain's typed objects. In async code, await pyactuator.invoke(...); the sync facade is for synchronous callers and also works from inside a running event loop.

from pyactuator import list_operations
from pyactuator.sync import invoke

# Every operation name, e.g. "pyactuator.trading.place_order",
# "pyactuator.google.sheets_append", "pyactuator.comms.slack_send" ...
print(list_operations())

result = invoke(
    "pyactuator.trading.place_order",
    {
        "symbol": "AAPL",
        "side": "buy",
        "quantity": "10",
        "order_type": "market",
        "client_order_id": "my-order-001",  # stable idempotency key
    },
)
print(result["success"], result["outcome"])  # True SUCCESS
print(result["adapter"]["real"])  # False — a mock served this call

Every envelope carries an adapter block, so a mock success is never mistaken for a real one. Pass require_real_adapters=True to refuse to run against a mock (or set PYACTUATOR_STRICT_ACTUATION=1 to make that the deployment default), or a configured registry= to actuate against real credentials.

Pass a policy_enforcer= to decide before the adapter is called. A denied call returns error.code: "POLICY_DENIED"; a dry-run decision returns dry_run: true with success: false, because nothing happened. pyactuator.runner.run and register_pyactuator_actions accept the same argument.

from decimal import Decimal

from pyactuator.core import LimitNotional
from pyactuator.sync import invoke

result = invoke(
    "pyactuator.trading.place_order",
    {
        "symbol": "AAPL",
        "side": "buy",
        "quantity": "100",
        "order_type": "limit",
        "limit_price": "250",
        "client_order_id": "my-order-002",
    },
    policy_enforcer=LimitNotional(Decimal("10000")),
)
print(result["success"], result["error"]["code"])  # False POLICY_DENIED

Quick start: Trading (mock broker)

import asyncio
from decimal import Decimal

from pyactuator import OrderRequest, Side, OrderType, TimeInForce
from pyactuator.domains.trading.adapters.mock import MockExecutionClient


async def main() -> None:
    client = MockExecutionClient()

    response = await client.submit(OrderRequest(
        client_order_id="my-order-001",
        symbol="AAPL",
        side=Side.BUY,
        quantity=Decimal("10"),
        order_type=OrderType.MARKET,
        time_in_force=TimeInForce.DAY,
    ))
    print(response.success, response.external_order_id)

    status = await client.get_status(response.external_order_id)
    print(status.status)  # ExecutionStatus.FILLED

    await client.close()


asyncio.run(main())

Quick start: Google Sheets (mock)

import asyncio

from pyactuator.domains.google.types import AppendRowsRequest
from pyactuator.domains.google.adapters.mock import MockSheetsClient


async def main() -> None:
    sheets = MockSheetsClient()
    # create_spreadsheet takes the title and returns the new spreadsheet id.
    spreadsheet_id = await sheets.create_spreadsheet(title="Daily PnL")

    response = await sheets.append_rows(AppendRowsRequest(
        spreadsheet_id=spreadsheet_id,
        range="Sheet1!A1",
        values=[["2025-05-07", "AAPL", 1234.56]],
    ))
    print(spreadsheet_id, response.success)


asyncio.run(main())

Quick start: PyStator action

PyActuator's PyStator bridge installs every domain operation as a kwargs-aware action callable on a pystator.actions.ActionRegistry. State machines call them by name from YAML, with parameters templated from the FSM context.

import asyncio

from pystator.actions import ActionRegistry, ActionExecutor
from pystator.actions.types import ActionSpec

from pyactuator.core.audit import MemoryAuditWriter
from pyactuator.core.registry import AdapterRegistry
from pyactuator.domains.trading.adapters.mock import MockExecutionClient
from pyactuator.integrations.pystator import register_pyactuator_actions


async def main() -> None:
    adapters = AdapterRegistry()
    adapters.register("trading", "broker", MockExecutionClient())

    audit = MemoryAuditWriter()
    registry = ActionRegistry()

    register_pyactuator_actions(registry, adapters=adapters, audit_writer=audit)

    executor = ActionExecutor(registry)
    result = await executor.execute_action_spec(
        ActionSpec(
            name="pyactuator.trading.place_order",
            params={
                "symbol": "AAPL",
                "side": "buy",
                "quantity": 1,
                "order_type": "market",
                "client_order_id": "coid-1",
            },
        ),
        ctx={},
    )
    print(result["success"], result["result"]["external_order_id"])
    print("audit records:", len(audit.records()))


asyncio.run(main())

Quick start: PyGubernator tool

from pyactuator.core.registry import AdapterRegistry
from pyactuator.domains.trading.adapters.mock import MockExecutionClient
from pyactuator.integrations.pygubernator import (
    iter_pyactuator_tool_specs,
    register_pyactuator_tools,
)
from pygubernator.tools import ToolRegistry


adapters = AdapterRegistry()
adapters.register("trading", "broker", MockExecutionClient())

tools = ToolRegistry()
register_pyactuator_tools(tools, adapters=adapters)

for spec in iter_pyactuator_tool_specs():
    print(spec.name, spec.metadata.side_effects, spec.metadata.idempotent)

Architecture

┌────────────────────────────────────────────────────────────┐
│   Callers                                                  │
│   ─────────────────────────────────────────────────        │
│   PyStator FSMs   PyGubernator agents   HTTP clients       │
└────────────┬──────────────┬──────────────┬─────────────────┘
             │              │              │
        actions          tools          /api/v1/<domain>
             │              │              │
┌────────────▼──────────────▼──────────────▼─────────────────┐
│                    pyactuator                              │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  core: envelopes • idempotency • retry • audit       │  │
│  │        policy • errors • registry • spec             │  │
│  └──────────────────────────────────────────────────────┘  │
│  ┌────────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐  │
│  │ trading    │ │ google   │ │ comms    │ │ documents   │  │
│  │ Alpaca/Mock│ │Drive/Sheets│ │Slack/Email│ │PDF/MD/HTML │  │
│  └────────────┘ └──────────┘ └──────────┘ └─────────────┘  │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  integrations: pystator • pygubernator               │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘
                       │
              ┌────────┴────────┬────────────┬───────────┐
              ▼                 ▼            ▼           ▼
          Broker API      Google APIs    Slack API    SMTP / SendGrid
        (Alpaca, IB)    (Drive/Sheets)

See ARCHITECTURE.md for the full layer model and docs/guides/ for the per-topic guides:

• domains-overview.md — what domains ship today and which surface to call them through.
• adding-a-domain.md — step-by-step recipe for adding a new domain.
• bridges.md — PyStator and PyGubernator bridge usage.

Domains shipped today

Trading: error hierarchy

The trading domain still exposes its full error hierarchy from the package root (re-exported from pyactuator.domains.trading.errors):

PyActuatorError
├── OrderValidationError     # bad OrderRequest fields
├── OrderRejectedError       # broker explicitly rejected
├── OrderNotFoundError       # get_status / cancel on unknown order
├── BrokerConnectionError    # network / timeout (retryable)
└── BrokerAdapterError       # adapter misconfiguration

from pyactuator import (
    PyActuatorError,
    OrderNotFoundError,
    BrokerConnectionError,
)

try:
    status = await client.get_status("unknown-id")
except OrderNotFoundError:
    print("Order does not exist at broker")
except BrokerConnectionError:
    print("Transient failure — safe to retry")
except PyActuatorError:
    print("Catch-all for any pyactuator error")

Adapter registry

All bridges and the FastAPI app share a single pyactuator.core.registry.AdapterRegistry keyed on (domain, kind):

from pyactuator.core.registry import AdapterRegistry
from pyactuator.domains.trading.adapters.mock import MockExecutionClient
from pyactuator.domains.comms.adapters.mock import MockSlackClient

adapters = AdapterRegistry()
adapters.register("trading", "broker", MockExecutionClient())
adapters.register("comms", "slack", MockSlackClient())

broker = adapters.resolve("trading", "broker")
slack = adapters.resolve("comms", "slack")

Swapping a real adapter for a mock at startup is a one-liner.

Audit

Every bridge call writes an AuditRecord through an injected AuditWriter:

from pyactuator.core.audit import MemoryAuditWriter

audit = MemoryAuditWriter()
register_pyactuator_actions(action_registry, adapters=adapters, audit_writer=audit)

for record in audit.records():
    print(record.kind, record.action, record.success)

NullAuditWriter is the default; production deployments wire their own writer (e.g. backed by a database or a log shipper).

HTTP API

Each domain ships a FastAPI router under /api/v1/<domain>/...:

Adapters resolve from app.state.adapter_registry. See pyactuator.api.main for the wiring and pyactuator.api.dependencies.* for the resolvers.

API reference (trading)

Protocols

Types

Enums

Adapters

License

MIT.