pyactuator 0.0.6

Multi-domain actuation core: trading, Google Workspace, comms, documents — plus PyStator and PyGubernator bridges

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), and Documents (PDF / Markdown / HTML) are first-class siblings.

Features

• Multi-domain — pyactuator.domains.{trading,google,comms,documents}, each with the same shape: types, protocols, errors, adapters/.
• 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.
• 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, reportlab are all optional extras. The core has zero mandatory dependencies.
• 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)
pip install pyactuator

# 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]

# 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: 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()
    await sheets.create_spreadsheet(spreadsheet_id="sheet-1", title="Daily PnL")

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


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

Domain	Operations	Adapters
trading	submit, get_status, cancel, subscribe_fills, get_positions, get_account	MockExecutionClient, AlpacaExecutionClient
google	Sheets append/read, Drive upload, Docs replace text / create	Mock + HTTPX-backed real adapters
comms	Slack send (webhook / API), Email send (SMTP / SendGrid)	Mock + real adapters
documents	Render templated PDF / Markdown / HTML	MockDocumentRenderer, JinjaMarkdownRenderer, ReportlabPdfRenderer

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>/...:

Route	Purpose
POST /api/v1/orders	Submit an order
GET /api/v1/orders/{id}	Order status
POST /api/v1/google/sheets/append	Append rows to a sheet
POST /api/v1/google/drive/upload	Upload a file to Drive
POST /api/v1/google/docs/replace-text	Replace text in a Google Doc
POST /api/v1/comms/slack/send	Send a Slack message
POST /api/v1/comms/email/send	Send an email
POST /api/v1/documents/render	Render a templated document

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

Protocol	Methods
ExecutionClient	submit, get_status, cancel, subscribe_fills, close
BrokerQueryClient	get_positions, get_position, get_account

Types

Type	Purpose
OrderRequest	Submit order params (validates on construction)
OrderResponse	Submission result (success, external_order_id, status)
OrderStatus	Full order status from broker
Fill	Execution/fill report
CancelResponse	Cancellation result
Position	Current position for a symbol
Account	Brokerage account info

Enums

Enum	Values
Side	BUY, SELL
OrderType	MARKET, LIMIT, STOP, STOP_LIMIT
TimeInForce	DAY, GTC, IOC, FOK, OPG, CLS
ExecutionStatus	PENDING_NEW, OPEN, PARTIALLY_FILLED, FILLED, CANCELED, REJECTED, EXPIRED

Adapters

Adapter	Install	Protocols
MockExecutionClient	core	ExecutionClient + BrokerQueryClient
AlpacaExecutionClient	pyactuator[trading-alpaca]	ExecutionClient + BrokerQueryClient
RetryExecutionClient	core	wraps any ExecutionClient with retry

License

MIT.