Deployment

Canary deployment support for safe strategy rollouts.

CanaryDeployment

almanak.framework.deployment.CanaryDeployment

CanaryDeployment(
    strategy_id: str,
    stable_version_id: str,
    canary_version_id: str,
    config: CanaryConfig | None = None,
    on_promote: CanaryCallback | None = None,
    on_rollback: CanaryCallback | None = None,
    on_state_change: CanaryCallback | None = None,
    metrics_provider: MetricsProvider | None = None,
    chain: str = "unknown",
)

Manages canary deployments for safe strategy version testing.

This class handles the full lifecycle of a canary deployment: 1. Deploy canary with limited capital allocation 2. Run both versions in parallel 3. Monitor and compare performance 4. Auto-promote or auto-rollback based on criteria

Attributes:

Name	Type	Description
strategy_id		ID of the strategy being deployed
stable_version_id		ID of the current stable version
canary_version_id		ID of the new canary version
state		Current deployment state
config		Deployment configuration

Initialize the canary deployment manager.

Parameters:

Name	Type	Description	Default
strategy_id	str	ID of the strategy being deployed	required
stable_version_id	str	ID of the current stable version	required
canary_version_id	str	ID of the new canary version	required
config	CanaryConfig | None	Deployment configuration (uses defaults if not provided)	None
on_promote	CanaryCallback | None	Callback when canary is promoted	None
on_rollback	CanaryCallback | None	Callback when canary is rolled back	None
on_state_change	CanaryCallback | None	Callback on any state change	None
metrics_provider	MetricsProvider | None	Function to fetch metrics for a version	None
chain	str	Blockchain network for event emission	'unknown'

deploy_canary async

deploy_canary(capital_usd: Decimal) -> DeployCanaryResult

Start the canary deployment.

Allocates capital between canary and stable versions and begins the observation period.

Parameters:

Name	Type	Description	Default
capital_usd	Decimal	Total capital to allocate across both versions	required

Returns:

Type	Description
DeployCanaryResult	DeployCanaryResult with deployment details

compare_performance

compare_performance() -> CanaryComparison

Compare canary performance against stable version.

Evaluates performance metrics against promotion criteria and returns a decision recommendation.

Returns:

Type	Description
CanaryComparison	CanaryComparison with metrics and decision

promote_canary async

promote_canary() -> CanaryResult

Promote the canary version to stable.

Makes the canary version the new stable version and reallocates all capital to it.

Returns:

Type	Description
CanaryResult	CanaryResult indicating success or failure

rollback_canary async

rollback_canary() -> CanaryResult

Rollback the canary deployment.

Stops the canary version and reallocates all capital back to the stable version.

Returns:

Type	Description
CanaryResult	CanaryResult indicating success or failure

cancel_deployment async

cancel_deployment() -> CanaryResult

Cancel the canary deployment without promoting or rolling back.

Returns:

Type	Description
CanaryResult	CanaryResult indicating success or failure

update_canary_metrics

update_canary_metrics(
    pnl_usd: Decimal | None = None,
    trades: int | None = None,
    errors: int | None = None,
    drawdown: Decimal | None = None,
) -> None

Manually update canary metrics.

Used when metrics_provider is not available.

Parameters:

Name	Type	Description	Default
pnl_usd	Decimal | None	Net PnL in USD	None
trades	int | None	Number of trades	None
errors	int | None	Number of errors	None
drawdown	Decimal | None	Max drawdown	None

update_stable_metrics

update_stable_metrics(
    pnl_usd: Decimal | None = None,
    trades: int | None = None,
    errors: int | None = None,
    drawdown: Decimal | None = None,
) -> None

Manually update stable metrics.

Used when metrics_provider is not available.

Parameters:

Name	Type	Description	Default
pnl_usd	Decimal | None	Net PnL in USD	None
trades	int | None	Number of trades	None
errors	int | None	Number of errors	None
drawdown	Decimal | None	Max drawdown	None

get_status

get_status() -> dict[str, Any]

Get the current status summary.

Returns:

Type	Description
dict[str, Any]	Dictionary with deployment status

to_dict

to_dict() -> dict[str, Any]

Export the deployment state for persistence.

Returns:

Type	Description
dict[str, Any]	Dictionary containing full deployment state

from_dict classmethod

from_dict(
    data: dict[str, Any],
    on_promote: CanaryCallback | None = None,
    on_rollback: CanaryCallback | None = None,
    on_state_change: CanaryCallback | None = None,
    metrics_provider: MetricsProvider | None = None,
) -> CanaryDeployment

Restore a deployment from persisted state.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary with deployment data	required
on_promote	CanaryCallback | None	Optional promotion callback	None
on_rollback	CanaryCallback | None	Optional rollback callback	None
on_state_change	CanaryCallback | None	Optional state change callback	None
metrics_provider	MetricsProvider | None	Optional metrics provider function	None

Returns:

Type	Description
CanaryDeployment	CanaryDeployment instance with restored state

CanaryConfig

almanak.framework.deployment.CanaryConfig dataclass

CanaryConfig(
    canary_percent: int = 10,
    observation_period_minutes: int = 60,
    auto_promote: bool = True,
    auto_rollback: bool = True,
    promotion_criteria: PromotionCriteria = PromotionCriteria(),
    check_interval_seconds: int = 60,
    emit_events: bool = True,
)

Configuration for a canary deployment.

Attributes:

Name	Type	Description
canary_percent	int	Percentage of total capital allocated to canary (1-50)
observation_period_minutes	int	How long to observe before deciding (min 5)
auto_promote	bool	Automatically promote if criteria met
auto_rollback	bool	Automatically rollback if criteria not met
promotion_criteria	PromotionCriteria	Criteria for promotion decisions
check_interval_seconds	int	How often to check metrics
emit_events	bool	Whether to emit timeline events

__post_init__

__post_init__() -> None

Validate configuration after initialization.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

from_dict classmethod

from_dict(data: dict[str, Any]) -> CanaryConfig

Create from dictionary.

CanaryState

almanak.framework.deployment.CanaryState dataclass

CanaryState(
    deployment_id: str,
    strategy_id: str,
    stable_version_id: str,
    canary_version_id: str,
    status: CanaryStatus = CanaryStatus.PENDING,
    config: CanaryConfig = CanaryConfig(),
    started_at: datetime | None = None,
    ended_at: datetime | None = None,
    canary_metrics: CanaryMetrics | None = None,
    stable_metrics: CanaryMetrics | None = None,
    total_capital_usd: Decimal = Decimal("0"),
    decision_history: list[dict[str, Any]] = list(),
)

Current state of a canary deployment.

Tracks the full state of an ongoing or completed canary deployment.

Attributes:

Name	Type	Description
deployment_id	str	Unique identifier for this deployment
strategy_id	str	Strategy being deployed
stable_version_id	str	ID of the stable version
canary_version_id	str	ID of the canary version
status	CanaryStatus	Current deployment status
config	CanaryConfig	Deployment configuration
started_at	datetime | None	When the deployment started
ended_at	datetime | None	When the deployment ended (if complete)
canary_metrics	CanaryMetrics | None	Performance metrics for canary
stable_metrics	CanaryMetrics | None	Performance metrics for stable
total_capital_usd	Decimal	Total capital across both versions
decision_history	list[dict[str, Any]]	History of decisions made

canary_capital_usd property

canary_capital_usd: Decimal

Calculate capital allocated to canary.

stable_capital_usd property

stable_capital_usd: Decimal

Calculate capital allocated to stable.

observation_deadline property

observation_deadline: datetime | None

Calculate when observation period ends.

observation_remaining_seconds property

observation_remaining_seconds: int

Calculate seconds remaining in observation period.

observation_complete property

observation_complete: bool

Check if observation period is complete.

__post_init__

__post_init__() -> None

Generate deployment ID if not provided.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

from_dict classmethod

from_dict(data: dict[str, Any]) -> CanaryState

Create from dictionary.

CanaryStatus

almanak.framework.deployment.CanaryStatus

Bases: str, Enum

Status of a canary deployment.

CanaryMetrics

almanak.framework.deployment.CanaryMetrics dataclass

CanaryMetrics(
    version_id: str,
    capital_allocated_usd: Decimal,
    metrics: PerformanceMetrics,
    error_count: int = 0,
    trade_count: int = 0,
    is_canary: bool = False,
    measurement_start: datetime | None = None,
)

Performance metrics tracked during canary deployment.

Extends PerformanceMetrics with canary-specific tracking.

Attributes:

Name	Type	Description
version_id	str	The version these metrics belong to
capital_allocated_usd	Decimal	Capital allocated to this version
metrics	PerformanceMetrics	The underlying performance metrics
error_count	int	Number of errors encountered
trade_count	int	Number of trades executed
is_canary	bool	Whether this is the canary version
measurement_start	datetime | None	When metrics collection started

error_rate property

error_rate: Decimal

Calculate the error rate.

duration_seconds property

duration_seconds: int

Get the duration of metrics collection.

__post_init__

__post_init__() -> None

Set default values after initialization.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

from_dict classmethod

from_dict(data: dict[str, Any]) -> CanaryMetrics

Create from dictionary.

CanaryResult

almanak.framework.deployment.CanaryResult dataclass

CanaryResult(
    success: bool,
    decision: CanaryDecision = CanaryDecision.CONTINUE,
    comparison: CanaryComparison | None = None,
    error: str | None = None,
    message: str = "",
)

Result of a canary decision or action.

Attributes:

Name	Type	Description
success	bool	Whether the action succeeded
decision	CanaryDecision	The decision made
comparison	CanaryComparison | None	Performance comparison (if available)
error	str | None	Error message if failed
message	str	Human-readable status message

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

CanaryDecision

almanak.framework.deployment.CanaryDecision

Bases: str, Enum

Decision to make about a canary deployment.