Quant Resource Implementation Patterns

Guide for implementing domain resources that integrate with OptAIC's resource-based architecture.

When to Use

Apply when:

• Creating new domain resource types (Dataset, Signal, Portfolio, etc.)
• Implementing Definition resources (PipelineDef, StoreDef, OpDef, MLModuleDef, PortfolioOptimizerDef)
• Implementing Instance resources (DatasetInstance, SignalInstance, BacktestInstance, etc.)
• Implementing Run resources (BacktestRun, TrainingRun, MonitoringRun, etc.)
• Adding domain-specific DB models, DTOs, or services

Three-Tier Resource Model

OptAIC separates Definitions (plugins) from Instances (configs) from Runs (executions):

Definition (Plugin)         Instance (Config)              Run (Execution)
─────────────────          ──────────────────             ────────────────
BloombergPipelineDef   →   SPX_OHLCV_Dataset          →   PipelineRun (daily)
PortfolioOptimizerDef  →   MVO_Conservative           →   PortfolioOptimizationRun
MLModuleDef            →   XGBoost_Predictor          →   TrainingRun, InferenceRun, MonitoringRun
(none)                 →   BacktestInstance           →   BacktestRun

Definitions: Reusable building blocks submitted as plugins (the "Law") Instances: Concrete configurations referencing definitions Runs: Executions that produce immutable versions and metrics

Resource Type Summary

Definition Resources

Instance Resources

Run Resources

Implementation Workflow

1. Determine Resource Tier

Definition resource? → Implements abstract interface, has test suite, requires evaluation Instance resource? → References definition(s), has config, can be scheduled

2. Create DB Model

Location: libs/db/models/<domain>.py

Link to resources table via FK. See [references/db-patterns.md](references/db-patterns.md).

3. Create DTOs

Location: libs/core/domain/<domain>.py

Use Pydantic. Never expose SQLAlchemy models to API. See [references/dto-patterns.md](references/dto-patterns.md).

4. Create Service Layer

Location: libs/core/domain/<domain>_service.py

Emit activities for all mutations. See [references/service-patterns.md](references/service-patterns.md).

5. Register ResourceType

Update libs/core/resources.py → ResourceType enum.

6. Generate Migration

optaic db revision --autogenerate -m "add <domain> resource"

7. Write Tests

Location: libs/core/tests/test_<domain>.py

Critical Rules

1. Lazy imports - Heavy deps (pandas, numpy, torch) must use TYPE_CHECKING blocks
2. Activity emission - All mutations emit activities in service layer
3. Guardrails hooks - Validate at lifecycle gates (create/update/promote)
4. Version tracking - Instances reference definition versions
5. coderef linkage - Services bridge DB models to factories via Definition.coderef

code_ref Integration (CRITICAL)

The code_ref field in Definition extension tables links to factory registration keys:

Definition.code_ref → FACTORY.build(code_ref) → Execution Object

Pattern: Service loads Instance → loads Definition → gets code_ref → builds from Factory

See [Service Patterns](references/service-patterns.md) for implementation details.

API Router Patterns

Location: apps/api/routers/<domain>.py

Implemented Routers

Router Pattern

from apps.api.deps import get_actor, get_db
from apps.api.rbac_utils import authorize_or_403, get_resource_or_404
from apps.api.services import DatasetService

@router.post("/{id}/preview", response_model=DatasetPreviewOut)
async def preview_dataset(
    dataset_id: UUID,
    payload: DatasetPreviewRequest = Body(...),
    actor: ActorContext = Depends(get_actor),
    db: AsyncSession = Depends(get_db),
) -> DatasetPreviewOut:
    # 1. Get resource and check RBAC
    resource = await get_resource_or_404(db, actor.tenant_id, dataset_id)
    await authorize_or_403(db, actor, Permission.RESOURCE_READ, resource.id)

    # 2. Call service (services emit activities, NOT routers)
    service = DatasetService()
    result = await service.preview(session=db, actor=actor, ...)

    # 3. Return DTO (never SQLAlchemy models)
    return DatasetPreviewOut(**result)

Service Layer

Location: apps/api/services/<domain>_service.py

Implemented Services

Service Exports

# apps/api/services/__init__.py
from apps.api.services.dataset_service import DatasetService
from apps.api.services.experiment_service import ExperimentService
from apps.api.services.op_service import OpService
from apps.api.services.pipeline_service import PipelineService
from apps.api.services.signal_service import SignalService

API Schemas

Location: apps/api/schemas.py (Quant Domain section)

System Bootstrap and Space Hierarchy

System Bootstrap

On application startup, the system bootstraps via libs/core/bootstrap.py:

from libs.core.bootstrap import (
    SYSTEM_TENANT_ID,      # 00000000-0000-0000-0000-000000000001
    SYSTEM_SPACE_ID,       # 00000000-0000-0000-0000-000000000002
    SYSTEM_PRINCIPAL_ID,   # 00000000-0000-0000-0000-000000000003
    SYSTEM_TENANT_ROOT_ID, # 00000000-0000-0000-0000-000000000010
    SYSTEM_PROJECT_ID,     # 00000000-0000-0000-0000-000000000013
)

# bootstrap_system() creates (idempotent):
# 1. System Tenant
# 2. Admin Principal ([email protected])
# 3. TenantRoot Resource
# 4. Default role permissions (owner, operator, viewer, auditor)
# 5. System Space with Official + Staging sub-spaces
# 6. System Project for definitions
# 7. Admin owner role on System Space

Space Hierarchy

TenantRoot
└── Space (space_kind: personal|team|system)
    ├── Subspace (subspace_kind: official)  ← Production resources
    ├── Subspace (subspace_kind: staging)   ← Resources under review
    └── Subspace (subspace_kind: custom)    ← User-created
        └── Project
            └── Resources (datasets, experiments, etc.)

SpaceService Patterns

apps/api/services/space_service.py provides:

class SpaceService:
    async def create_space_with_subspaces(...) -> SpaceCreationResult
        # Creates Space + Official + Staging sub-spaces
        # Grants owner role to owner_principal_id
        # Emits activities for all creations

    async def create_user_with_personal_space(...) -> UserCreationResult
        # Creates Principal + Personal Space
        # Grants owner on Personal Space
        # (Optionally grant VIEW on System Space)

    async def create_team_space(...) -> SpaceCreationResult
        # Creates Team Space with owner
        # Optionally grants operator to members

Resource Copy Pattern

Copy definitions from System Space to user projects:

# API: POST /resources/{resource_id}/copy
# SDK: client.resources.copy(resource_id, target_parent_id, new_name=...)

# Creates:
# 1. New resource with copier as owner
# 2. derived_from lineage edge to source
# 3. resource.copied activity

Admin SDK

libs/sdk_py/admin.py provides admin operations:

# Create user with Personal Space
result = await client.admin.create_user_with_space(
    display_name="Alice Smith",
    email="[email protected]",
)
# Returns: principal_id, space_id, official_subspace_id, staging_subspace_id

# Create Team Space
result = await client.admin.create_team_space(
    name="Quant Research Team",
    owner_principal_id=owner_id,
    member_principal_ids=[member1, member2],  # Optional
)

Reference Files

• [DB Model Patterns](references/db-patterns.md) - SQLAlchemy patterns
• [DTO Patterns](references/dto-patterns.md) - Pydantic schemas
• [Service Patterns](references/service-patterns.md) - CRUD with activities