Skip to content

Models

composer

Composer API Client.

ApplySubscription

Bases: StrEnum

Subscription type enum.

Source code in composer/models/backtest/requests.py 
28
29
30
31
32
33
class ApplySubscription(StrEnum):
    """Subscription type enum."""

    NONE = "none"
    MONTHLY = "monthly"
    YEARLY = "yearly"

Asset

Bases: BaseNode

A ticker/asset node representing a security to trade.

Source code in composer/models/common/symphony.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
class Asset(BaseNode):
    """A ticker/asset node representing a security to trade."""

    model_config = {"populate_by_name": True}

    step: Literal["asset"] = Field(default="asset")
    name: str | None = Field(None, description="Display name of the asset")
    ticker: str = Field(description="Ticker symbol (e.g., AAPL, CRYPTO::BTC//USD)")
    exchange: str | None = Field(None, description="Exchange code (e.g., XNAS)")
    price: float | None = Field(None, description="Current price (API-populated)")
    dollar_volume: float | None = Field(None, alias="dollar_volume")
    has_marketcap: bool | None = Field(None, alias="has_marketcap")
    children_count: int | None = Field(None, alias="children-count")

BacktestExistingSymphonyRequest

Bases: BacktestParams

Backtest request for an existing saved symphony.

Use this when backtesting a symphony that has already been saved. The symphony_id is provided in the URL path, not the request body.

Source code in composer/models/backtest/requests.py 
105
106
107
108
109
110
111
112
113
class BacktestExistingSymphonyRequest(BacktestParams):
    """
    Backtest request for an existing saved symphony.

    Use this when backtesting a symphony that has already been saved.
    The symphony_id is provided in the URL path, not the request body.
    """

    pass

BacktestRequest

Bases: BacktestParams

Full backtest request with symphony definition.

Use this when backtesting a new symphony by providing the full symphony definition in the request body.

Source code in composer/models/backtest/requests.py 
 94
 95
 96
 97
 98
 99
100
101
102
class BacktestRequest(BacktestParams):
    """
    Full backtest request with symphony definition.

    Use this when backtesting a new symphony by providing the full
    symphony definition in the request body.
    """

    symphony: SymphonyDefinition = Field(description="Symphony definition to backtest")

BacktestResult

Bases: BaseModel

Complete backtest result.

This is the main response model returned by both backtest endpoints: - POST /api/v0.1/backtest (backtest by definition) - POST /api/v0.1/symphonies/{symphony-id}/backtest (backtest existing symphony)

Contains performance metrics, holdings history, costs, and statistics.

Source code in composer/models/backtest/responses.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class BacktestResult(BaseModel):
    """
    Complete backtest result.

    This is the main response model returned by both backtest endpoints:
    - POST /api/v0.1/backtest (backtest by definition)
    - POST /api/v0.1/symphonies/{symphony-id}/backtest (backtest existing symphony)

    Contains performance metrics, holdings history, costs, and statistics.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    # Timing and metadata
    last_semantic_update_at: str | None = None
    sparkgraph_url: str | None = None
    first_day: str | None = None
    last_market_day: str | None = None

    # Daily Value Metrics (DVM)
    dvm_capital: _DateSeriesDict | None = None

    # Time-Dependent Value Metrics (TDVM) weights
    tdvm_weights: _DateSeriesDict | None = None

    @field_validator("dvm_capital", mode="before")
    @classmethod
    def transform_dvm_capital(cls, v):
        """Transform dvm_capital field to date-indexed dict."""
        transformed = _transform_dvm_to_by_date(v)
        if transformed is None:
            return None
        return _DateSeriesDict(transformed)

    @field_validator("tdvm_weights", mode="before")
    @classmethod
    def transform_tdvm_weights(cls, v):
        """Transform tdvm_weights field to date-indexed dict."""
        transformed = _transform_dvm_to_by_date(v)
        if transformed is None:
            return None
        result = _DateSeriesDict(transformed, fill_value=False)
        for d in result:
            for ticker in result[d]:
                result[d][ticker] = bool(result[d][ticker])
        return result

    @field_validator("first_day", mode="before")
    @classmethod
    def transform_first_day(cls, v):
        """Transform first_day from epoch to date string."""
        if v is None:
            return None
        return _epoch_day_to_date_string(v)

    @field_validator("last_market_day", mode="before")
    @classmethod
    def transform_last_market_day(cls, v):
        """Transform last_market_day from epoch to date string."""
        if v is None:
            return None
        return _epoch_day_to_date_string(v)

    @field_validator("rebalance_days", mode="before")
    @classmethod
    def transform_rebalance_days(cls, v):
        """Transform rebalance_days from epochs to date strings."""
        if v is None:
            return None
        return [_epoch_day_to_date_string(d) for d in v]

    # Rebalancing information
    rebalance_days: list[str] | None = None
    active_asset_nodes: dict[str, float] | None = None

    # Costs breakdown
    costs: Costs | None = None

    # Final state
    last_market_days_value: float | None = None
    last_market_days_holdings: dict[str, float] | None = None

    # Legend and warnings
    legend: dict[str, LegendEntry] | None = None
    data_warnings: dict[str, list[DataWarning]] | None = None
    benchmark_errors: list[str] | None = None

    # Performance statistics
    stats: Stats | None = None

    def __repr__(self) -> str:
        """Return string representation of BacktestResult."""
        parts = []
        if self.stats:
            parts.append(f"stats={self.stats}")
        if self.last_market_days_value is not None:
            parts.append(f"final_value={self.last_market_days_value:,.2f}")
        if self.first_day is not None and self.last_market_day is not None:
            first = datetime.strptime(self.first_day, "%Y-%m-%d").date()
            last = datetime.strptime(self.last_market_day, "%Y-%m-%d").date()
            parts.append(f"days={(last - first).days}")
        return f"BacktestResult({', '.join(parts)})"

    def __str__(self) -> str:
        """Return string representation of BacktestResult."""
        return self.__repr__()

__repr__()

Return string representation of BacktestResult.

Source code in composer/models/backtest/responses.py 
190
191
192
193
194
195
196
197
198
199
200
201
def __repr__(self) -> str:
    """Return string representation of BacktestResult."""
    parts = []
    if self.stats:
        parts.append(f"stats={self.stats}")
    if self.last_market_days_value is not None:
        parts.append(f"final_value={self.last_market_days_value:,.2f}")
    if self.first_day is not None and self.last_market_day is not None:
        first = datetime.strptime(self.first_day, "%Y-%m-%d").date()
        last = datetime.strptime(self.last_market_day, "%Y-%m-%d").date()
        parts.append(f"days={(last - first).days}")
    return f"BacktestResult({', '.join(parts)})"

__str__()

Return string representation of BacktestResult.

Source code in composer/models/backtest/responses.py 
203
204
205
def __str__(self) -> str:
    """Return string representation of BacktestResult."""
    return self.__repr__()

transform_dvm_capital(v) classmethod

Transform dvm_capital field to date-indexed dict.

Source code in composer/models/backtest/responses.py 
125
126
127
128
129
130
131
132
@field_validator("dvm_capital", mode="before")
@classmethod
def transform_dvm_capital(cls, v):
    """Transform dvm_capital field to date-indexed dict."""
    transformed = _transform_dvm_to_by_date(v)
    if transformed is None:
        return None
    return _DateSeriesDict(transformed)

transform_first_day(v) classmethod

Transform first_day from epoch to date string.

Source code in composer/models/backtest/responses.py 
147
148
149
150
151
152
153
@field_validator("first_day", mode="before")
@classmethod
def transform_first_day(cls, v):
    """Transform first_day from epoch to date string."""
    if v is None:
        return None
    return _epoch_day_to_date_string(v)

transform_last_market_day(v) classmethod

Transform last_market_day from epoch to date string.

Source code in composer/models/backtest/responses.py 
155
156
157
158
159
160
161
@field_validator("last_market_day", mode="before")
@classmethod
def transform_last_market_day(cls, v):
    """Transform last_market_day from epoch to date string."""
    if v is None:
        return None
    return _epoch_day_to_date_string(v)

transform_rebalance_days(v) classmethod

Transform rebalance_days from epochs to date strings.

Source code in composer/models/backtest/responses.py 
163
164
165
166
167
168
169
@field_validator("rebalance_days", mode="before")
@classmethod
def transform_rebalance_days(cls, v):
    """Transform rebalance_days from epochs to date strings."""
    if v is None:
        return None
    return [_epoch_day_to_date_string(d) for d in v]

transform_tdvm_weights(v) classmethod

Transform tdvm_weights field to date-indexed dict.

Source code in composer/models/backtest/responses.py 
134
135
136
137
138
139
140
141
142
143
144
145
@field_validator("tdvm_weights", mode="before")
@classmethod
def transform_tdvm_weights(cls, v):
    """Transform tdvm_weights field to date-indexed dict."""
    transformed = _transform_dvm_to_by_date(v)
    if transformed is None:
        return None
    result = _DateSeriesDict(transformed, fill_value=False)
    for d in result:
        for ticker in result[d]:
            result[d][ticker] = bool(result[d][ticker])
    return result

BacktestVersion

Bases: StrEnum

Backtest version enum.

Source code in composer/models/backtest/requests.py 
11
12
13
14
15
class BacktestVersion(StrEnum):
    """Backtest version enum."""

    V1 = "v1"
    V2 = "v2"

BaseNode

Bases: BaseModel

Base class for all symphony nodes.

Source code in composer/models/common/symphony.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class BaseNode(BaseModel):
    """Base class for all symphony nodes."""

    model_config = {"populate_by_name": True, "extra": "ignore"}

    id: str = Field(
        default_factory=lambda: str(uuid.uuid4()),
        description="Unique identifier for this node (auto-generated UUID or custom ID)",
    )
    weight: WeightMap | None = Field(None, description="Weight fraction for this node")

    @field_validator("id")
    @classmethod
    def validate_id(cls, v):
        """Ensure ID is a valid UUID. If not, auto-generate one."""
        try:
            uuid.UUID(v)
            return v
        except (ValueError, AttributeError):
            # Not a valid UUID, generate a new one
            return str(uuid.uuid4())

    def model_dump(self, **kwargs) -> dict[str, Any]:
        """Override model_dump to exclude None values by default."""
        kwargs.setdefault("exclude_none", True)
        return super().model_dump(**kwargs)

    def model_dump_json(self, **kwargs) -> str:
        """Override model_dump_json to exclude None values by default."""
        kwargs.setdefault("exclude_none", True)
        return super().model_dump_json(**kwargs)

    @model_validator(mode="after")
    def parse_children(self):
        """Recursively parse children dicts into typed models."""
        return self

model_dump(**kwargs)

Override model_dump to exclude None values by default.

Source code in composer/models/common/symphony.py 
140
141
142
143
def model_dump(self, **kwargs) -> dict[str, Any]:
    """Override model_dump to exclude None values by default."""
    kwargs.setdefault("exclude_none", True)
    return super().model_dump(**kwargs)

model_dump_json(**kwargs)

Override model_dump_json to exclude None values by default.

Source code in composer/models/common/symphony.py 
145
146
147
148
def model_dump_json(self, **kwargs) -> str:
    """Override model_dump_json to exclude None values by default."""
    kwargs.setdefault("exclude_none", True)
    return super().model_dump_json(**kwargs)

parse_children()

Recursively parse children dicts into typed models.

Source code in composer/models/common/symphony.py 
150
151
152
153
@model_validator(mode="after")
def parse_children(self):
    """Recursively parse children dicts into typed models."""
    return self

validate_id(v) classmethod

Ensure ID is a valid UUID. If not, auto-generate one.

Source code in composer/models/common/symphony.py 
129
130
131
132
133
134
135
136
137
138
@field_validator("id")
@classmethod
def validate_id(cls, v):
    """Ensure ID is a valid UUID. If not, auto-generate one."""
    try:
        uuid.UUID(v)
        return v
    except (ValueError, AttributeError):
        # Not a valid UUID, generate a new one
        return str(uuid.uuid4())

BenchmarkStats

Bases: BaseModel

Statistics for a benchmark comparison.

Source code in composer/models/common/stats.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class BenchmarkStats(BaseModel):
    """Statistics for a benchmark comparison."""

    calmar_ratio: float | None = None
    cumulative_return: float | None = None
    max_: float | None = Field(None, alias="max")
    max_drawdown: float | None = None
    mean: float | None = None
    median: float | None = None
    min_: float | None = Field(None, alias="min")
    skewness: float | None = None
    kurtosis: float | None = None
    sharpe_ratio: float | None = None
    sortino_ratio: float | None = None
    size: int | None = None
    standard_deviation: float | None = None
    annualized_rate_of_return: float | None = None
    annualized_turnover: float | None = None
    trailing_one_day_return: float | None = None
    trailing_one_week_return: float | None = None
    trailing_two_week_return: float | None = None
    trailing_one_month_return: float | None = None
    trailing_three_month_return: float | None = None
    trailing_one_year_return: float | None = None
    top_one_day_contribution: float | None = None
    top_five_percent_day_contribution: float | None = None
    top_ten_percent_day_contribution: float | None = None
    herfindahl_index: float | None = None
    win_rate: float | None = None
    tail_ratio: float | None = None
    percent: RegressionMetrics | None = None
    log: RegressionMetrics | None = None

    class Config:
        """Pydantic model configuration."""

        populate_by_name = True

Config

Pydantic model configuration.

Source code in composer/models/common/stats.py 
48
49
50
51
class Config:
    """Pydantic model configuration."""

    populate_by_name = True

Broker

Bases: StrEnum

Broker enum.

Source code in composer/models/backtest/requests.py 
18
19
20
21
22
23
24
25
class Broker(StrEnum):
    """Broker enum."""

    ALPACA_OAUTH = "ALPACA_OAUTH"
    ALPACA_WHITE_LABEL = "ALPACA_WHITE_LABEL"
    APEX_LEGACY = "APEX_LEGACY"
    ALPACA = "alpaca"
    APEX = "apex"

ComposerClient

Main client for interacting with the Composer API.

This client provides access to various Composer API resources including backtest, market data, trading, portfolio management, and more.

Parameters:

Name Type Description Default
api_key str

The API Key ID

 required
api_secret str

The API Secret Key

 required
timeout float

Request timeout in seconds (default: 30.0)

 30.0
retry_config RetryConfig | None

Configuration for retry behavior (default: 3 retries, 10s for 429, 3s for 5xx)

 _DEFAULT_RETRY_CONFIG
Source code in composer/client.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class ComposerClient:
    """Main client for interacting with the Composer API.

    This client provides access to various Composer API resources including
    backtest, market data, trading, portfolio management, and more.

    Args:
        api_key: The API Key ID
        api_secret: The API Secret Key
        timeout: Request timeout in seconds (default: 30.0)
        retry_config: Configuration for retry behavior (default: 3 retries, 10s for 429, 3s for 5xx)
    """

    def __init__(
        self,
        api_key: str,
        api_secret: str,
        timeout: float = 30.0,
        retry_config: RetryConfig | None = _DEFAULT_RETRY_CONFIG,
    ):
        """
        Initialize the Composer Client.

        Args:
            api_key (str): The API Key ID
            api_secret (str): The API Secret Key
            timeout (float): Request timeout in seconds (default: 30.0)
            retry_config (RetryConfig | None): Configuration for retry behavior.
                Pass None to disable retries (default: RetryConfig())
        """
        self.http_client = HTTPClient(
            api_key, api_secret, timeout=timeout, retry_config=retry_config
        )
        # Separate HTTP clients for backtest API (different base URL)
        # Public client (no auth headers) for public endpoints
        # Authenticated client for endpoints requiring auth
        self.backtest_auth_client = HTTPClient(
            api_key,
            api_secret,
            base_url="https://backtest-api.composer.trade/",
            timeout=timeout,
            retry_config=retry_config,
        )
        # Stagehand API client for portfolio/account endpoints
        self.stagehand_auth_client = HTTPClient(
            api_key,
            api_secret,
            base_url="https://stagehand-api.composer.trade/",
            timeout=timeout,
            retry_config=retry_config,
        )
        # Trading API client for deploy and trading endpoints
        self.trading_auth_client = HTTPClient(
            api_key,
            api_secret,
            base_url="https://trading-api.composer.trade/",
            timeout=timeout,
            retry_config=retry_config,
        )

        # Initialize resources
        self.backtest = Backtest(self.backtest_auth_client)
        # Market data uses stagehand API
        self.market_data = MarketData(self.stagehand_auth_client)
        # Accounts, Portfolio, Cash, User, AI Agents, Conversation, Reports, Auth use stagehand API
        self.accounts = Accounts(self.stagehand_auth_client)
        self.portfolio = Portfolio(self.stagehand_auth_client)
        self.cash = Cash(self.stagehand_auth_client)
        self.user = User(self.stagehand_auth_client)
        self.ai_agents = AIAgents(self.stagehand_auth_client)
        self.conversation = Conversation(self.stagehand_auth_client)
        self.reports = Reports(self.stagehand_auth_client)
        self.auth_management = AuthManagement(self.stagehand_auth_client)
        self.search = Search(self.backtest_auth_client)
        # Quotes uses stagehand API (public endpoint)
        self.quotes = Quotes(self.stagehand_auth_client)

        # Public user endpoints
        self.public_user = PublicUser(self.stagehand_auth_client)

        # New backtest API resources
        # Public endpoints (no auth required)
        self.public_symphony = PublicSymphony(self.backtest_auth_client)
        # Authenticated endpoints
        self.user_symphony = UserSymphony(self.backtest_auth_client)
        self.user_symphonies = UserSymphonies(self.backtest_auth_client)
        self.watchlist = Watchlist(self.backtest_auth_client)
        # Trading API resources (deploy, dry-run, and trading endpoints)
        self.deploy = DeployResource(self.trading_auth_client)
        self.dry_run = DryRun(self.trading_auth_client)
        self.trading = Trading(self.trading_auth_client)

__init__(api_key, api_secret, timeout=30.0, retry_config=_DEFAULT_RETRY_CONFIG)

Initialize the Composer Client.

Parameters:

Name Type Description Default
api_key str

The API Key ID

 required
api_secret str

The API Secret Key

 required
timeout float

Request timeout in seconds (default: 30.0)

 30.0
retry_config RetryConfig | None

Configuration for retry behavior. Pass None to disable retries (default: RetryConfig())

 _DEFAULT_RETRY_CONFIG
Source code in composer/client.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def __init__(
    self,
    api_key: str,
    api_secret: str,
    timeout: float = 30.0,
    retry_config: RetryConfig | None = _DEFAULT_RETRY_CONFIG,
):
    """
    Initialize the Composer Client.

    Args:
        api_key (str): The API Key ID
        api_secret (str): The API Secret Key
        timeout (float): Request timeout in seconds (default: 30.0)
        retry_config (RetryConfig | None): Configuration for retry behavior.
            Pass None to disable retries (default: RetryConfig())
    """
    self.http_client = HTTPClient(
        api_key, api_secret, timeout=timeout, retry_config=retry_config
    )
    # Separate HTTP clients for backtest API (different base URL)
    # Public client (no auth headers) for public endpoints
    # Authenticated client for endpoints requiring auth
    self.backtest_auth_client = HTTPClient(
        api_key,
        api_secret,
        base_url="https://backtest-api.composer.trade/",
        timeout=timeout,
        retry_config=retry_config,
    )
    # Stagehand API client for portfolio/account endpoints
    self.stagehand_auth_client = HTTPClient(
        api_key,
        api_secret,
        base_url="https://stagehand-api.composer.trade/",
        timeout=timeout,
        retry_config=retry_config,
    )
    # Trading API client for deploy and trading endpoints
    self.trading_auth_client = HTTPClient(
        api_key,
        api_secret,
        base_url="https://trading-api.composer.trade/",
        timeout=timeout,
        retry_config=retry_config,
    )

    # Initialize resources
    self.backtest = Backtest(self.backtest_auth_client)
    # Market data uses stagehand API
    self.market_data = MarketData(self.stagehand_auth_client)
    # Accounts, Portfolio, Cash, User, AI Agents, Conversation, Reports, Auth use stagehand API
    self.accounts = Accounts(self.stagehand_auth_client)
    self.portfolio = Portfolio(self.stagehand_auth_client)
    self.cash = Cash(self.stagehand_auth_client)
    self.user = User(self.stagehand_auth_client)
    self.ai_agents = AIAgents(self.stagehand_auth_client)
    self.conversation = Conversation(self.stagehand_auth_client)
    self.reports = Reports(self.stagehand_auth_client)
    self.auth_management = AuthManagement(self.stagehand_auth_client)
    self.search = Search(self.backtest_auth_client)
    # Quotes uses stagehand API (public endpoint)
    self.quotes = Quotes(self.stagehand_auth_client)

    # Public user endpoints
    self.public_user = PublicUser(self.stagehand_auth_client)

    # New backtest API resources
    # Public endpoints (no auth required)
    self.public_symphony = PublicSymphony(self.backtest_auth_client)
    # Authenticated endpoints
    self.user_symphony = UserSymphony(self.backtest_auth_client)
    self.user_symphonies = UserSymphonies(self.backtest_auth_client)
    self.watchlist = Watchlist(self.backtest_auth_client)
    # Trading API resources (deploy, dry-run, and trading endpoints)
    self.deploy = DeployResource(self.trading_auth_client)
    self.dry_run = DryRun(self.trading_auth_client)
    self.trading = Trading(self.trading_auth_client)

Costs

Bases: BaseModel

Fee and cost breakdown from a backtest.

Shows all the costs incurred during the backtest including regulatory fees, TAF fees, slippage, spread markup, and subscription costs.

Source code in composer/models/backtest/responses.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
class Costs(BaseModel):
    """
    Fee and cost breakdown from a backtest.

    Shows all the costs incurred during the backtest including
    regulatory fees, TAF fees, slippage, spread markup, and subscription costs.
    """

    reg_fee: float
    taf_fee: float
    slippage: float
    spread_markup: float
    subscription: float

DataWarning

Bases: BaseModel

Data quality warning for a specific ticker.

Source code in composer/models/backtest/responses.py 
86
87
88
89
90
91
class DataWarning(BaseModel):
    """Data quality warning for a specific ticker."""

    message: str
    recommended_start_date: str | None = None
    recommended_end_date: str | None = None

Empty

Bases: BaseNode

Empty/cash placeholder node.

Source code in composer/models/common/symphony.py 
174
175
176
177
178
179
class Empty(BaseNode):
    """Empty/cash placeholder node."""

    model_config = {"populate_by_name": True}

    step: Literal["empty"] = Field(default="empty")

Filter

Bases: BaseNode

Filter node for selecting top/bottom N assets.

Source code in composer/models/common/symphony.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class Filter(BaseNode):
    """Filter node for selecting top/bottom N assets."""

    model_config = {"populate_by_name": True}

    step: Literal["filter"] = Field(default="filter")
    select_: bool | None = Field(None, alias="select?")
    select_fn: Literal["top", "bottom"] | None = Field(None, alias="select-fn")
    select_n: int | None = Field(None, alias="select-n")
    sort_by_: bool | None = Field(None, alias="sort-by?")
    sort_by_fn: Function | None = Field(None, alias="sort-by-fn")
    sort_by_fn_params: dict[str, Any] | None = Field(None, alias="sort-by-fn-params")
    sort_by_window_days: int | None = Field(None, alias="sort-by-window-days")
    children: list[Any] = Field(default_factory=list)

Function

Bases: StrEnum

Mathematical and technical analysis functions available for conditions and filters.

Source code in composer/models/common/symphony.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class Function(StrEnum):
    """Mathematical and technical analysis functions available for conditions and filters."""

    CUMULATIVE_RETURN = "cumulative-return"
    CURRENT_PRICE = "current-price"
    EXPONENTIAL_MOVING_AVERAGE_PRICE = "exponential-moving-average-price"
    LOWER_BOLLINGER = "lower-bollinger"
    MAX_DRAWDOWN = "max-drawdown"
    MOVING_AVERAGE_PRICE = "moving-average-price"
    MOVING_AVERAGE_CONVERGENCE_DIVERGENCE = "moving-average-convergence-divergence"
    MOVING_AVERAGE_CONVERGENCE_DIVERGENCE_SIGNAL = "moving-average-convergence-divergence-signal"
    MOVING_AVERAGE_RETURN = "moving-average-return"
    PERCENTAGE_PRICE_OSCILLATOR = "percentage-price-oscillator"
    PERCENTAGE_PRICE_OSCILLATOR_SIGNAL = "percentage-price-oscillator-signal"
    RELATIVE_STRENGTH_INDEX = "relative-strength-index"
    STANDARD_DEVIATION_PRICE = "standard-deviation-price"
    STANDARD_DEVIATION_RETURN = "standard-deviation-return"
    UPPER_BOLLINGER = "upper-bollinger"

Group

Bases: BaseNode

Group node containing a single weight strategy.

Source code in composer/models/common/symphony.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class Group(BaseNode):
    """Group node containing a single weight strategy."""

    model_config = {"populate_by_name": True}

    step: Literal["group"] = Field(default="group")
    name: str | None = Field(None)
    collapsed: bool | None = Field(None, alias="collapsed?")
    children: list[Any] = Field(default_factory=list)

    @field_validator("children")
    @classmethod
    def validate_single_child(cls, v):
        """Validate that group has exactly one child."""
        if len(v) != 1:
            raise ValueError("Group must have exactly one child")
        return v

validate_single_child(v) classmethod

Validate that group has exactly one child.

Source code in composer/models/common/symphony.py 
289
290
291
292
293
294
295
@field_validator("children")
@classmethod
def validate_single_child(cls, v):
    """Validate that group has exactly one child."""
    if len(v) != 1:
        raise ValueError("Group must have exactly one child")
    return v

If

Bases: BaseNode

If/Else conditional node.

Source code in composer/models/common/symphony.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
class If(BaseNode):
    """If/Else conditional node."""

    model_config = {"populate_by_name": True}

    step: Literal["if"] = Field(default="if")
    children: list[IfChildTrue | IfChildFalse] = Field(default_factory=list)

    @field_validator("children")
    @classmethod
    def validate_if_children(cls, v):
        """Ensure If node has exactly one true and one false child."""
        if len(v) != 2:
            raise ValueError("If node must have exactly 2 children")

        true_count = sum(1 for child in v if not child.is_else_condition)
        false_count = sum(1 for child in v if child.is_else_condition)

        if true_count != 1 or false_count != 1:
            raise ValueError("If node must have one true and one false condition")

        return v

validate_if_children(v) classmethod

Ensure If node has exactly one true and one false child.

Source code in composer/models/common/symphony.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@field_validator("children")
@classmethod
def validate_if_children(cls, v):
    """Ensure If node has exactly one true and one false child."""
    if len(v) != 2:
        raise ValueError("If node must have exactly 2 children")

    true_count = sum(1 for child in v if not child.is_else_condition)
    false_count = sum(1 for child in v if child.is_else_condition)

    if true_count != 1 or false_count != 1:
        raise ValueError("If node must have one true and one false condition")

    return v

IfChildFalse

Bases: BaseNode

The 'false' (else) branch of an If condition.

Source code in composer/models/common/symphony.py 
207
208
209
210
211
212
213
214
class IfChildFalse(BaseNode):
    """The 'false' (else) branch of an If condition."""

    model_config = {"populate_by_name": True}

    step: Literal["if-child"] = Field(default="if-child")
    is_else_condition: Literal[True] | None = Field(True, alias="is-else-condition?")
    children: list[Any] = Field(default_factory=list)

IfChildTrue

Bases: BaseNode

The 'true' branch of an If condition.

Source code in composer/models/common/symphony.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class IfChildTrue(BaseNode):
    """The 'true' branch of an If condition."""

    model_config = {"populate_by_name": True}

    step: Literal["if-child"] = Field(default="if-child")
    is_else_condition: Literal[False] | None = Field(False, alias="is-else-condition?")
    comparator: Literal["gt", "gte", "eq", "lt", "lte"] | None = Field(
        None, description="Comparison operator"
    )
    lhs_fn: Function | None = Field(None, alias="lhs-fn")
    lhs_window_days: int | None = Field(None, alias="lhs-window-days")
    lhs_val: str | None = Field(None, alias="lhs-val")
    lhs_fn_params: dict[str, Any] | None = Field(None, alias="lhs-fn-params")
    rhs_val: str | float | None = Field(None, alias="rhs-val")
    rhs_fixed_value: bool | None = Field(None, alias="rhs-fixed-value?")
    rhs_fn: Function | None = Field(None, alias="rhs-fn")
    rhs_window_days: int | None = Field(None, alias="rhs-window-days")
    rhs_fn_params: dict[str, Any] | None = Field(None, alias="rhs-fn-params")
    children: list[Any] = Field(default_factory=list)

Quote

Bases: BaseModel

Quote for a single ticker in rebalance.

Source code in composer/models/backtest/requests.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class Quote(BaseModel):
    """Quote for a single ticker in rebalance."""

    ticker: str
    trading_halted: bool = False
    open: float
    close: float | None = None
    low: float
    high: float
    volume: int
    bid: dict[str, Any] | None = None
    ask: dict[str, Any] | None = None
    last: dict[str, Any] | None = None
    source: str | None = None
    timestamp: str | None = None

RebalanceFrequency

Bases: StrEnum

How often the symphony should rebalance its holdings.

Source code in composer/models/common/symphony.py 
78
79
80
81
82
83
84
85
86
class RebalanceFrequency(StrEnum):
    """How often the symphony should rebalance its holdings."""

    NONE = "none"
    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"
    QUARTERLY = "quarterly"
    YEARLY = "yearly"

RebalanceRequest

Bases: BaseModel

Request body for the rebalance endpoint.

Used to run a rebalance for specified symphonies given a starting state.

Source code in composer/models/backtest/requests.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
class RebalanceRequest(BaseModel):
    """
    Request body for the rebalance endpoint.

    Used to run a rebalance for specified symphonies given a starting state.
    """

    dry_run: bool = False
    broker: Broker = Broker.ALPACA_WHITE_LABEL
    adjust_for_dtbp: bool = False
    disable_fractional_trading: bool = False
    fractionability: dict[str, bool] | None = None
    end_date: str | None = None
    quotes: dict[str, Quote] | None = None
    symphonies: dict[str, SymphonyRebalanceState]

RebalanceResult

Bases: BaseModel

Result from a rebalance request.

Contains quotes, fractionability info, and run results for each symphony.

Source code in composer/models/backtest/responses.py 
227
228
229
230
231
232
233
234
235
236
237
class RebalanceResult(BaseModel):
    """
    Result from a rebalance request.

    Contains quotes, fractionability info, and run results for each symphony.
    """

    quotes: dict[str, Any] | None = None
    fractionability: dict[str, bool] | None = None
    adjusted_for_dtbp: bool = False
    run_results: dict[str, SymphonyRunResult] | None = None

RecommendedTrade

Bases: BaseModel

A recommended trade in a dry run or preview.

Source code in composer/models/dry_run/__init__.py 
15
16
17
18
19
20
21
22
23
24
25
class RecommendedTrade(BaseModel):
    """A recommended trade in a dry run or preview."""

    model_config = {"populate_by_name": True}

    ticker: str
    notional: float
    quantity: float
    prev_value: float
    prev_weight: float
    next_weight: float

RegressionMetrics

Bases: BaseModel

Regression metrics for benchmark comparison (alpha, beta, etc.).

Source code in composer/models/common/stats.py 
 6
 7
 8
 9
10
11
12
class RegressionMetrics(BaseModel):
    """Regression metrics for benchmark comparison (alpha, beta, etc.)."""

    alpha: float | None = None
    beta: float | None = None
    r_square: float | None = Field(None, alias="r_square")
    pearson_r: float | None = None

RetryConfig

Bases: BaseModel

Configuration for retry behavior.

Source code in composer/http_client.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class RetryConfig(BaseModel):
    """Configuration for retry behavior."""

    max_retries: int = Field(default=3, description="Maximum number of retry attempts")
    rate_limit_wait: float = Field(
        default=10.0, description="Initial seconds to wait on 429 responses"
    )
    server_error_wait: float = Field(
        default=3.0, description="Initial seconds to wait on 500, 502, 503, 504 responses"
    )
    exponential_base: float = Field(
        default=2.0, description="Base for exponential backoff multiplier"
    )
    retry_statuses: set[int] = Field(
        default={429, 500, 502, 503, 504}, description="HTTP status codes that trigger a retry"
    )

Stats

Bases: BaseModel

Performance statistics for portfolios, backtests, and symphonies.

Contains all key performance metrics like Sharpe ratio, cumulative return, max drawdown, and various trailing returns.

Source code in composer/models/common/stats.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class Stats(BaseModel):
    """
    Performance statistics for portfolios, backtests, and symphonies.

    Contains all key performance metrics like Sharpe ratio, cumulative return,
    max drawdown, and various trailing returns.
    """

    calmar_ratio: float | None = None
    cumulative_return: float | None = None
    max_: float | None = Field(None, alias="max")
    max_drawdown: float | None = None
    mean: float | None = None
    median: float | None = None
    min_: float | None = Field(None, alias="min")
    skewness: float | None = None
    kurtosis: float | None = None
    sharpe_ratio: float | None = None
    sortino_ratio: float | None = None
    size: int | None = None
    standard_deviation: float | None = None
    annualized_rate_of_return: float | None = None
    annualized_turnover: float | None = None
    trailing_one_day_return: float | None = None
    trailing_one_week_return: float | None = None
    trailing_two_week_return: float | None = None
    trailing_one_month_return: float | None = None
    trailing_three_month_return: float | None = None
    trailing_one_year_return: float | None = None
    top_one_day_contribution: float | None = None
    top_five_percent_day_contribution: float | None = None
    top_ten_percent_day_contribution: float | None = None
    herfindahl_index: float | None = None
    win_rate: float | None = None
    tail_ratio: float | None = None
    benchmarks: dict[str, BenchmarkStats] | None = None

    class Config:
        """Pydantic model configuration."""

        populate_by_name = True

    def __repr__(self) -> str:
        """Return string representation of Stats."""
        parts = []
        if self.sharpe_ratio is not None:
            parts.append(f"sharpe={self.sharpe_ratio:.2f}")
        if self.cumulative_return is not None:
            parts.append(f"cumulative={self.cumulative_return:.2%}")
        if self.max_drawdown is not None:
            parts.append(f"drawdown={self.max_drawdown:.2%}")
        if self.annualized_rate_of_return is not None:
            parts.append(f"ann_return={self.annualized_rate_of_return:.2%}")
        return f"Stats({', '.join(parts)})"

    def __str__(self) -> str:
        """Return string representation of Stats."""
        return self.__repr__()
        return self.__repr__()

Config

Pydantic model configuration.

Source code in composer/models/common/stats.py 
91
92
93
94
class Config:
    """Pydantic model configuration."""

    populate_by_name = True

__repr__()

Return string representation of Stats.

Source code in composer/models/common/stats.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __repr__(self) -> str:
    """Return string representation of Stats."""
    parts = []
    if self.sharpe_ratio is not None:
        parts.append(f"sharpe={self.sharpe_ratio:.2f}")
    if self.cumulative_return is not None:
        parts.append(f"cumulative={self.cumulative_return:.2%}")
    if self.max_drawdown is not None:
        parts.append(f"drawdown={self.max_drawdown:.2%}")
    if self.annualized_rate_of_return is not None:
        parts.append(f"ann_return={self.annualized_rate_of_return:.2%}")
    return f"Stats({', '.join(parts)})"

__str__()

Return string representation of Stats.

Source code in composer/models/common/stats.py 
109
110
111
112
def __str__(self) -> str:
    """Return string representation of Stats."""
    return self.__repr__()
    return self.__repr__()

SymphonyDefinition

Bases: BaseNode

Definition of a symphony/strategy.

Source code in composer/models/common/symphony.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
class SymphonyDefinition(BaseNode):
    """Definition of a symphony/strategy."""

    model_config = {"populate_by_name": True}

    step: Literal["root"] = Field(default="root")
    name: str = Field(description="Display name of the symphony")
    description: str = Field(default="", description="Description of the strategy")
    rebalance: RebalanceFrequency = Field(description="Rebalancing frequency")
    rebalance_corridor_width: float | None = Field(None, alias="rebalance-corridor-width")
    children: list[WeightCashEqual | WeightCashSpecified | WeightInverseVol] = Field(
        default_factory=list
    )

    @model_validator(mode="after")
    def parse_children(self):
        """Recursively parse children dicts into typed models."""
        if self.children:
            self.children = _parse_node(self.children)
        return self

    @field_validator("rebalance_corridor_width")
    @classmethod
    def validate_corridor_width(cls, v, info):
        """Corridor width only allowed when rebalance is 'none'."""
        rebalance = info.data.get("rebalance")
        if v is not None and rebalance != "none":
            raise ValueError('rebalance_corridor_width can only be set when rebalance is "none"')
        return v

    @field_validator("children")
    @classmethod
    def validate_single_child(cls, v):
        """SymphonyDefinition must have exactly one weight child."""
        if len(v) != 1:
            raise ValueError("SymphonyDefinition must have exactly one child")
        return v

parse_children()

Recursively parse children dicts into typed models.

Source code in composer/models/common/symphony.py 
339
340
341
342
343
344
@model_validator(mode="after")
def parse_children(self):
    """Recursively parse children dicts into typed models."""
    if self.children:
        self.children = _parse_node(self.children)
    return self

validate_corridor_width(v, info) classmethod

Corridor width only allowed when rebalance is 'none'.

Source code in composer/models/common/symphony.py 
346
347
348
349
350
351
352
353
@field_validator("rebalance_corridor_width")
@classmethod
def validate_corridor_width(cls, v, info):
    """Corridor width only allowed when rebalance is 'none'."""
    rebalance = info.data.get("rebalance")
    if v is not None and rebalance != "none":
        raise ValueError('rebalance_corridor_width can only be set when rebalance is "none"')
    return v

validate_single_child(v) classmethod

SymphonyDefinition must have exactly one weight child.

Source code in composer/models/common/symphony.py 
355
356
357
358
359
360
361
@field_validator("children")
@classmethod
def validate_single_child(cls, v):
    """SymphonyDefinition must have exactly one weight child."""
    if len(v) != 1:
        raise ValueError("SymphonyDefinition must have exactly one child")
    return v

SymphonyRebalanceState

Bases: BaseModel

State of a single symphony for rebalance.

Source code in composer/models/backtest/requests.py 
133
134
135
136
137
138
139
class SymphonyRebalanceState(BaseModel):
    """State of a single symphony for rebalance."""

    last_rebalanced_on: str | None = None
    cash: float
    unsettled_cash: float = 0.0
    shares: dict[str, float]

SymphonyRunResult

Bases: BaseModel

Result of running a single symphony during rebalance.

Source code in composer/models/backtest/responses.py 
218
219
220
221
222
223
224
class SymphonyRunResult(BaseModel):
    """Result of running a single symphony during rebalance."""

    next_rebalanced_after: str | None = None
    rebalanced: bool = False
    active_asset_nodes: dict[str, float] | None = None
    recommended_trades: list[RecommendedTrade] | None = None

UpdateSymphonyNodesResponse

Bases: BaseModel

Response from updating symphony nodes.

Source code in composer/models/symphony/__init__.py 
103
104
105
106
107
108
109
class UpdateSymphonyNodesResponse(BaseModel):
    """Response from updating symphony nodes."""

    model_config = {"populate_by_name": True}

    symphony_id: str = Field(description="ID of the symphony")
    version_id: str = Field(description="ID of the new version")

UpdateSymphonyResponse

Bases: BaseModel

Response from updating a symphony.

Source code in composer/models/symphony/__init__.py 
 94
 95
 96
 97
 98
 99
100
class UpdateSymphonyResponse(BaseModel):
    """Response from updating a symphony."""

    model_config = {"populate_by_name": True}

    existing_version_id: str = Field(description="ID of the previous version")
    version_id: str | None = Field(None, description="ID of the new version")

WeightCashEqual

Bases: BaseNode

Equal weighting across all children.

Source code in composer/models/common/symphony.py 
301
302
303
304
305
306
307
class WeightCashEqual(BaseNode):
    """Equal weighting across all children."""

    model_config = {"populate_by_name": True}

    step: Literal["wt-cash-equal"] | None = Field("wt-cash-equal")
    children: list[Any] = Field(default_factory=list)

WeightCashSpecified

Bases: BaseNode

Specified weighting with explicit weights.

Source code in composer/models/common/symphony.py 
313
314
315
316
317
318
319
class WeightCashSpecified(BaseNode):
    """Specified weighting with explicit weights."""

    model_config = {"populate_by_name": True}

    step: Literal["wt-cash-specified"] | None = Field("wt-cash-specified")
    children: list[Any] = Field(default_factory=list)

WeightInverseVol

Bases: BaseNode

Inverse volatility weighting.

Source code in composer/models/common/symphony.py 
266
267
268
269
270
271
272
273
class WeightInverseVol(BaseNode):
    """Inverse volatility weighting."""

    model_config = {"populate_by_name": True}

    step: Literal["wt-inverse-vol"] | None = Field("wt-inverse-vol")
    window_days: int | None = Field(None, alias="window-days")
    children: list[Any] = Field(default_factory=list)

WeightMap

Bases: BaseModel

Weight fraction represented as numerator/denominator.

Examples
- 50% = num: 50, den: 100
- 33.3% = num: 33.3, den: 100
Source code in composer/models/common/symphony.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class WeightMap(BaseModel):
    """
    Weight fraction represented as numerator/denominator.

    Examples
    --------
        - 50% = num: 50, den: 100
        - 33.3% = num: 33.3, den: 100
    """

    model_config = {"populate_by_name": True}

    num: str | int | float = Field(description="Numerator of the weight fraction")
    den: str | int | float = Field(description="Denominator of the weight fraction (typically 100)")

    @field_validator("num", "den")
    @classmethod
    def validate_positive(cls, v):
        """Ensure weight values are non-negative."""
        val = float(v) if isinstance(v, str) else v
        if val < 0:
            raise ValueError("Weight numerator and denominator must be non-negative")
        return v

validate_positive(v) classmethod

Ensure weight values are non-negative.

Source code in composer/models/common/symphony.py 
104
105
106
107
108
109
110
111
@field_validator("num", "den")
@classmethod
def validate_positive(cls, v):
    """Ensure weight values are non-negative."""
    val = float(v) if isinstance(v, str) else v
    if val < 0:
        raise ValueError("Weight numerator and denominator must be non-negative")
    return v

validate_symphony_score(score)

Validate a symphony score and check crypto rebalancing rules.

Parameters:

Name Type Description Default
score SymphonyDefinition | dict

Symphony score as SymphonyDefinition model or dict

 required
Returns
Validated SymphonyDefinition model
Source code in composer/models/common/symphony.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def validate_symphony_score(score: SymphonyDefinition | dict) -> SymphonyDefinition:
    """
    Validate a symphony score and check crypto rebalancing rules.

    Args:
        score: Symphony score as SymphonyDefinition model or dict

    Returns
    -------
        Validated SymphonyDefinition model
    """
    if isinstance(score, dict):
        score = SymphonyDefinition.model_validate(score)

    # Check for crypto assets
    crypto_tickers = []

    def find_crypto(node):
        """Recursively find crypto assets."""
        if isinstance(node, Asset) and node.ticker and node.ticker.startswith("CRYPTO::"):
            crypto_tickers.append(node.ticker)
        # Check children recursively for nodes that have children
        elif not isinstance(node, Asset) and hasattr(node, "children") and node.children:
            for child in node.children:
                if isinstance(child, BaseModel):
                    find_crypto(child)

    if score.children:
        find_crypto(score.children[0])

    # Validate crypto rebalancing
    if crypto_tickers and score.rebalance not in ["none", "daily"]:
        raise ValueError(
            f"Symphonies with crypto must use daily or threshold rebalancing. "
            f"Found rebalance={score.rebalance}"
        )

    return score