Backtest & Symphonies

Backtest Resource

composer.resources.backtest.Backtest

Backtest resource for running backtest simulations.

This resource provides methods for running backtests, rebalances, and managing backtest configurations.

Source code in composer/resources/backtest.py

class Backtest:
    """Backtest resource for running backtest simulations.

    This resource provides methods for running backtests, rebalances,
    and managing backtest configurations.
    """

    def __init__(self, http_client):
        self.http_client = http_client

    def run(self, request: BacktestRequest | dict[str, Any]) -> BacktestResult:
        """
        Run a generic backtest simulation (v1).

        Args:
            request: BacktestRequest model or dictionary matching the schema.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics

        Example:
            # Backtest with full symphony definition
            result = client.backtest.run(
                BacktestRequest(
                    symphony=SymphonyDefinition(
                        name="My Strategy",
                        rebalance="daily",
                        children=[...]
                    )
                )
            )
        """
        payload = request
        if isinstance(request, BacktestRequest):
            payload = request.model_dump(by_alias=True, exclude_none=True, mode="json")
            if "symphony" in payload and payload["symphony"]:
                payload["symphony"] = {"raw_value": payload["symphony"]}

        raw_response = self.http_client.post("/api/v1/backtest", json=payload)
        return BacktestResult.model_validate(raw_response)

    def run_v2(
        self,
        symphony: SymphonyDefinition | dict[str, Any] | None = None,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """
        Run a generic backtest simulation (v2).

        Args:
            symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics

        Example:
             result = client.backtest.run_v2(symphony=SymphonyDefinition(...))
             print(f"Sharpe: {result.stats.sharpe_ratio}")

        Example with dict:
             score_data = {"step": "root", "name": "My Strategy", ...}
             result = client.backtest.run_v2(symphony=score_data)
        """
        if symphony is not None:
            if isinstance(symphony, dict):
                raw_value = symphony
            else:
                raw_value = symphony.model_dump(by_alias=True, exclude_none=True)
            symphony_payload = {"raw_value": raw_value}
        else:
            symphony_payload = None

        payload = {
            "symphony": symphony_payload,
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self.http_client.post("/api/v2/backtest", json=payload)
        return BacktestResult.model_validate(raw_response)

    def run_public_v2(
        self,
        symphony: SymphonyDefinition | dict[str, Any] | None = None,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """
        Run a public backtest simulation (v2).

        Args:
            symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics

        Example:
             result = client.backtest.run_public_v2(symphony=SymphonyDefinition(...))
             print(f"Sharpe: {result.stats.sharpe_ratio}")

        Example with dict:
             score_data = {"step": "root", "name": "My Strategy", ...}
             result = client.backtest.run_public_v2(symphony=score_data)
        """
        if symphony is not None:
            if isinstance(symphony, dict):
                raw_value = symphony
            else:
                raw_value = symphony.model_dump(by_alias=True, exclude_none=True)
            symphony_payload = {"raw_value": raw_value}
        else:
            symphony_payload = None

        payload = {
            "symphony": symphony_payload,
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self.http_client.post("/api/v2/public/backtest", json=payload)
        return BacktestResult.model_validate(raw_response)

    def rebalance(self, request: RebalanceRequest | dict[str, Any]) -> RebalanceResult:
        """
        Run a rebalance for specified symphonies.

        Args:
            request: RebalanceRequest model or dictionary matching the schema.

        Returns
        -------
            RebalanceResult: Rebalance result with quotes and run results.

        Example:
             request = RebalanceRequest(
            ...     symphonies={
            ...         "sym-123": SymphonyRebalanceState(cash=10000, shares={})
            ...     }
            ... )
             result = client.backtest.rebalance(request)
             print(result.run_results)
        """
        payload = request
        if isinstance(request, RebalanceRequest):
            payload = request.model_dump(by_alias=True, exclude_none=True, mode="json")

        raw_response = self.http_client.post("/api/v2/rebalance", json=payload)
        return RebalanceResult.model_validate(raw_response)

    def list_configs(self, keys: list[str] | None = None) -> list[ConfigEntry]:
        """
        List accessible config entries.

        Args:
            keys: Optional list of config keys to filter by.
                Valid keys: "constants", "deposit_presets_config", "openai-prompt",
                "openai_config"

        Returns
        -------
            List of ConfigEntry objects

        Example:
            configs = client.backtest.list_configs()
            for cfg in configs:
                print(f"{cfg.config_key}: {cfg.config}")
        """
        params = {"keys": keys} if keys else None
        raw_response = self.http_client.get("/api/v1/configs", params=params)
        return [ConfigEntry.from_dict(c) for c in raw_response]

    def get_config(self, config_key: ConfigKey | str) -> ConfigEntry:
        """
        Get a single config entry by key.

        Args:
            config_key: The config key to retrieve.
                Valid keys: "constants", "deposit_presets_config", "openai-prompt",
                "openai_config"
                Can pass ConfigKey enum or string.

        Returns
        -------
            ConfigEntry object

        Example:
            config = client.backtest.get_config("openai_config")
            print(config.config_key)
            print(config.config.system_prompt)
        """
        key_value = config_key.value if isinstance(config_key, ConfigKey) else config_key
        raw_response = self.http_client.get(f"/api/v1/configs/{key_value}")
        return ConfigEntry.from_dict(raw_response)

get_config(config_key)

Get a single config entry by key.

Parameters:

Name	Type	Description	Default
config_key	ConfigKey | str	The config key to retrieve. Valid keys: "constants", "deposit_presets_config", "openai-prompt", "openai_config" Can pass ConfigKey enum or string.	required

Returns

ConfigEntry object

Example

config = client.backtest.get_config("openai_config") print(config.config_key) print(config.config.system_prompt)

Source code in composer/resources/backtest.py

def get_config(self, config_key: ConfigKey | str) -> ConfigEntry:
    """
    Get a single config entry by key.

    Args:
        config_key: The config key to retrieve.
            Valid keys: "constants", "deposit_presets_config", "openai-prompt",
            "openai_config"
            Can pass ConfigKey enum or string.

    Returns
    -------
        ConfigEntry object

    Example:
        config = client.backtest.get_config("openai_config")
        print(config.config_key)
        print(config.config.system_prompt)
    """
    key_value = config_key.value if isinstance(config_key, ConfigKey) else config_key
    raw_response = self.http_client.get(f"/api/v1/configs/{key_value}")
    return ConfigEntry.from_dict(raw_response)

list_configs(keys=None)

List accessible config entries.

Parameters:

Name	Type	Description	Default
keys	list[str] | None	Optional list of config keys to filter by. Valid keys: "constants", "deposit_presets_config", "openai-prompt", "openai_config"	None

Returns

List of ConfigEntry objects

Example

configs = client.backtest.list_configs() for cfg in configs: print(f"{cfg.config_key}: {cfg.config}")

Source code in composer/resources/backtest.py

def list_configs(self, keys: list[str] | None = None) -> list[ConfigEntry]:
    """
    List accessible config entries.

    Args:
        keys: Optional list of config keys to filter by.
            Valid keys: "constants", "deposit_presets_config", "openai-prompt",
            "openai_config"

    Returns
    -------
        List of ConfigEntry objects

    Example:
        configs = client.backtest.list_configs()
        for cfg in configs:
            print(f"{cfg.config_key}: {cfg.config}")
    """
    params = {"keys": keys} if keys else None
    raw_response = self.http_client.get("/api/v1/configs", params=params)
    return [ConfigEntry.from_dict(c) for c in raw_response]

rebalance(request)

Run a rebalance for specified symphonies.

Parameters:

Name	Type	Description	Default
request	RebalanceRequest | dict[str, Any]	RebalanceRequest model or dictionary matching the schema.	required

Returns

RebalanceResult: Rebalance result with quotes and run results.

Example

request = RebalanceRequest(

...     symphonies={
...         "sym-123": SymphonyRebalanceState(cash=10000, shares={})
...     }
... )
 result = client.backtest.rebalance(request)
 print(result.run_results)

Source code in composer/resources/backtest.py

def rebalance(self, request: RebalanceRequest | dict[str, Any]) -> RebalanceResult:
    """
    Run a rebalance for specified symphonies.

    Args:
        request: RebalanceRequest model or dictionary matching the schema.

    Returns
    -------
        RebalanceResult: Rebalance result with quotes and run results.

    Example:
         request = RebalanceRequest(
        ...     symphonies={
        ...         "sym-123": SymphonyRebalanceState(cash=10000, shares={})
        ...     }
        ... )
         result = client.backtest.rebalance(request)
         print(result.run_results)
    """
    payload = request
    if isinstance(request, RebalanceRequest):
        payload = request.model_dump(by_alias=True, exclude_none=True, mode="json")

    raw_response = self.http_client.post("/api/v2/rebalance", json=payload)
    return RebalanceResult.model_validate(raw_response)

run(request)

Run a generic backtest simulation (v1).

Parameters:

Name	Type	Description	Default
request	BacktestRequest | dict[str, Any]	BacktestRequest model or dictionary matching the schema.	required

Returns

BacktestResult: Parsed backtest result with all statistics

Example

Backtest with full symphony definition

result = client.backtest.run( BacktestRequest( symphony=SymphonyDefinition( name="My Strategy", rebalance="daily", children=[...] ) ) )

Source code in composer/resources/backtest.py

def run(self, request: BacktestRequest | dict[str, Any]) -> BacktestResult:
    """
    Run a generic backtest simulation (v1).

    Args:
        request: BacktestRequest model or dictionary matching the schema.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics

    Example:
        # Backtest with full symphony definition
        result = client.backtest.run(
            BacktestRequest(
                symphony=SymphonyDefinition(
                    name="My Strategy",
                    rebalance="daily",
                    children=[...]
                )
            )
        )
    """
    payload = request
    if isinstance(request, BacktestRequest):
        payload = request.model_dump(by_alias=True, exclude_none=True, mode="json")
        if "symphony" in payload and payload["symphony"]:
            payload["symphony"] = {"raw_value": payload["symphony"]}

    raw_response = self.http_client.post("/api/v1/backtest", json=payload)
    return BacktestResult.model_validate(raw_response)

run_public_v2(symphony=None, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a public backtest simulation (v2).

Parameters:

Name	Type	Description	Default
symphony	SymphonyDefinition | dict[str, Any] | None	Symphony definition to backtest (SymphonyDefinition model or dict).	None
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics

Example

result = client.backtest.run_public_v2(symphony=SymphonyDefinition(...)) print(f"Sharpe: {result.stats.sharpe_ratio}")

Example with dict

score_data = {"step": "root", "name": "My Strategy", ...} result = client.backtest.run_public_v2(symphony=score_data)

Source code in composer/resources/backtest.py

def run_public_v2(
    self,
    symphony: SymphonyDefinition | dict[str, Any] | None = None,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """
    Run a public backtest simulation (v2).

    Args:
        symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics

    Example:
         result = client.backtest.run_public_v2(symphony=SymphonyDefinition(...))
         print(f"Sharpe: {result.stats.sharpe_ratio}")

    Example with dict:
         score_data = {"step": "root", "name": "My Strategy", ...}
         result = client.backtest.run_public_v2(symphony=score_data)
    """
    if symphony is not None:
        if isinstance(symphony, dict):
            raw_value = symphony
        else:
            raw_value = symphony.model_dump(by_alias=True, exclude_none=True)
        symphony_payload = {"raw_value": raw_value}
    else:
        symphony_payload = None

    payload = {
        "symphony": symphony_payload,
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self.http_client.post("/api/v2/public/backtest", json=payload)
    return BacktestResult.model_validate(raw_response)

run_v2(symphony=None, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a generic backtest simulation (v2).

Parameters:

Name	Type	Description	Default
symphony	SymphonyDefinition | dict[str, Any] | None	Symphony definition to backtest (SymphonyDefinition model or dict).	None
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics

Example

result = client.backtest.run_v2(symphony=SymphonyDefinition(...)) print(f"Sharpe: {result.stats.sharpe_ratio}")

Example with dict

score_data = {"step": "root", "name": "My Strategy", ...} result = client.backtest.run_v2(symphony=score_data)

Source code in composer/resources/backtest.py

def run_v2(
    self,
    symphony: SymphonyDefinition | dict[str, Any] | None = None,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """
    Run a generic backtest simulation (v2).

    Args:
        symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics

    Example:
         result = client.backtest.run_v2(symphony=SymphonyDefinition(...))
         print(f"Sharpe: {result.stats.sharpe_ratio}")

    Example with dict:
         score_data = {"step": "root", "name": "My Strategy", ...}
         result = client.backtest.run_v2(symphony=score_data)
    """
    if symphony is not None:
        if isinstance(symphony, dict):
            raw_value = symphony
        else:
            raw_value = symphony.model_dump(by_alias=True, exclude_none=True)
        symphony_payload = {"raw_value": raw_value}
    else:
        symphony_payload = None

    payload = {
        "symphony": symphony_payload,
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self.http_client.post("/api/v2/backtest", json=payload)
    return BacktestResult.model_validate(raw_response)

Public Symphony Resource

composer.resources.public_symphony.PublicSymphony

Public symphony endpoints (no authentication required).

These endpoints provide read-only access to publicly shared symphonies and technical indicators.

Source code in composer/resources/public_symphony.py

class PublicSymphony:
    """
    Public symphony endpoints (no authentication required).

    These endpoints provide read-only access to publicly shared symphonies
    and technical indicators.
    """

    def __init__(self, http_client: HTTPClient):
        self._client = http_client

    def get_indicators(self) -> list[Indicator]:
        """
        Get a list of all available technical indicators.

        Returns a list of indicators that can be used in symphony conditions,
        including moving averages, RSI, standard deviation, etc.

        Returns
        -------
            List[Indicator]: List of available technical indicators with their
                parameters and metadata.

        Example:
             indicators = client.public_symphony.get_indicators()
             for indicator in indicators:
            ...     print(f"{indicator.name}: {indicator.description}")
        """
        response = self._client.get("/api/v1/public/symphony-scores/indicators")
        return [Indicator.model_validate(item) for item in response]

    def get_symphony_meta(self, symphony_ids: list[str]) -> list[SymphonyMeta]:
        """
        Get metadata for symphonies by their IDs.

        Args:
            symphony_ids: List of symphony IDs to look up.

        Returns
        -------
            List[SymphonyMeta]: List of symphony metadata objects.

        Example:
             meta = client.public_symphony.get_symphony_meta(["sym-abc123"])
             for symphony in meta:
            ...     print(f"{symphony.name}: {symphony.rebalance_frequency}")
        """
        response = self._client.post(
            "/api/v1/public/meta/symphonies", json={"symphony_id": symphony_ids}
        )
        result = SymphonyMetaResponse.model_validate(response)
        return result.symphonies

    def get_symphony(self, symphony_id: str) -> SymphonyDetail:
        """
        Get detailed information about a symphony.

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            SymphonyDetail: Detailed symphony information including stats,
                backtest data, and metadata.

        Example:
             symphony = client.public_symphony.get_symphony("sym-abc123")
             print(f"Name: {symphony.name}")
             print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")
        """
        response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}")
        return SymphonyDetail.model_validate(response)

    def get_versions(self, symphony_id: str) -> list[SymphonyVersionInfo]:
        """
        Get all versions for a symphony.

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            List[SymphonyVersionInfo]: List of version information.

        Example:
             versions = client.public_symphony.get_versions("sym-abc123")
             for version in versions:
            ...     print(f"Version {version.version_id}: {version.created_at}")
        """
        response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/versions")
        return [SymphonyVersionInfo.model_validate(item) for item in response]

    def get_version_score(
        self, symphony_id: str, version_id: str, score_version: str = "v1"
    ) -> SymphonyDefinition:
        """
        Get an existing symphony version's EDN (score).

        Args:
            symphony_id: Unique identifier for the symphony.
            version_id: Unique identifier for the symphony version.
            score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

        Returns
        -------
            SymphonyDefinition: The symphony score/parsed EDN structure.

        Example:
             score = client.public_symphony.get_version_score(
            ...     "sym-abc123",
            ...     "ver-xyz789"
            ... )
             print(score.name)
             print(score.rebalance)
        """
        params = {"score_version": score_version}
        response = self._client.get(
            f"/api/v1/public/symphonies/{symphony_id}/versions/{version_id}/score",
            params=params,
        )
        return SymphonyDefinition.model_validate(response)

    def get_score(self, symphony_id: str, score_version: str = "v1") -> SymphonyDefinition:
        """
        Get an existing symphony's EDN (score).

        Args:
            symphony_id: Unique identifier for the symphony.
            score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

        Returns
        -------
            SymphonyDefinition: The symphony score/parsed EDN structure.

        Example:
             score = client.public_symphony.get_score("sym-abc123")
             print(score.name)
             print(score.rebalance)
        """
        params = {"score_version": score_version}
        response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/score", params=params)
        return SymphonyDefinition.model_validate(response)

    def get_tickers(self, symphony_id: str) -> list[str]:
        """
        Get all tickers used in a symphony.

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            List[str]: List of ticker symbols used in the symphony.

        Example:
             tickers = client.public_symphony.get_tickers("sym-abc123")
             print(f"Symphony uses: {', '.join(tickers)}")
        """
        response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/tickers")
        result = TickersResponse.model_validate(response)
        return result.tickers

    def backtest_symphony(
        self,
        symphony_id: str,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """
        Run a backtest for a public symphony by its ID.

        Args:
            symphony_id: The ID of the symphony to backtest.
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics.

        Example:
            result = client.public_symphony.backtest_symphony("sym-abc123")
            print(f"Sharpe: {result.stats.sharpe_ratio}")
        """
        payload = {
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self._client.post(
            f"/api/v1/public/symphonies/{symphony_id}/backtest",
            json=payload,
        )
        return BacktestResult.model_validate(raw_response)

    def backtest(
        self,
        symphony: SymphonyDefinition | dict[str, Any],
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """
        Run a standalone backtest with a custom symphony definition.

        Args:
            symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics.

        Example:
             result = client.public_symphony.backtest(symphony=SymphonyDefinition(...))
             print(f"Sharpe: {result.stats.sharpe_ratio}")

        Example with dict:
             score_data = {"step": "root", "name": "My Strategy", ...}
             result = client.public_symphony.backtest(symphony=score_data)
        """
        if isinstance(symphony, dict):
            raw_value = symphony
        else:
            raw_value = symphony.model_dump(by_alias=True, exclude_none=True)

        payload = {
            "symphony": {"raw_value": raw_value},
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self._client.post(
            "/api/v1/public/backtest",
            json=payload,
        )
        return BacktestResult.model_validate(raw_response)

    def backtest_symphony_v2(
        self,
        symphony_id: str,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """
        Run a backtest for a public symphony by its ID (v2).

        Args:
            symphony_id: The ID of the symphony to backtest.
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics.

        Example:
             result = client.public_symphony.backtest_symphony_v2("sym-abc123")
             print(f"Sharpe: {result.stats.sharpe_ratio}")
        """
        payload = {
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self._client.post(
            f"/api/v2/public/symphonies/{symphony_id}/backtest",
            json=payload,
        )
        return BacktestResult.model_validate(raw_response)

backtest(symphony, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a standalone backtest with a custom symphony definition.

Parameters:

Name	Type	Description	Default
symphony	SymphonyDefinition | dict[str, Any]	Symphony definition to backtest (SymphonyDefinition model or dict).	required
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics.

Example

result = client.public_symphony.backtest(symphony=SymphonyDefinition(...)) print(f"Sharpe: {result.stats.sharpe_ratio}")

Example with dict

score_data = {"step": "root", "name": "My Strategy", ...} result = client.public_symphony.backtest(symphony=score_data)

Source code in composer/resources/public_symphony.py

def backtest(
    self,
    symphony: SymphonyDefinition | dict[str, Any],
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """
    Run a standalone backtest with a custom symphony definition.

    Args:
        symphony: Symphony definition to backtest (SymphonyDefinition model or dict).
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics.

    Example:
         result = client.public_symphony.backtest(symphony=SymphonyDefinition(...))
         print(f"Sharpe: {result.stats.sharpe_ratio}")

    Example with dict:
         score_data = {"step": "root", "name": "My Strategy", ...}
         result = client.public_symphony.backtest(symphony=score_data)
    """
    if isinstance(symphony, dict):
        raw_value = symphony
    else:
        raw_value = symphony.model_dump(by_alias=True, exclude_none=True)

    payload = {
        "symphony": {"raw_value": raw_value},
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self._client.post(
        "/api/v1/public/backtest",
        json=payload,
    )
    return BacktestResult.model_validate(raw_response)

backtest_symphony(symphony_id, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a backtest for a public symphony by its ID.

Parameters:

Name	Type	Description	Default
symphony_id	str	The ID of the symphony to backtest.	required
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics.

Example

result = client.public_symphony.backtest_symphony("sym-abc123") print(f"Sharpe: {result.stats.sharpe_ratio}")

Source code in composer/resources/public_symphony.py

def backtest_symphony(
    self,
    symphony_id: str,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """
    Run a backtest for a public symphony by its ID.

    Args:
        symphony_id: The ID of the symphony to backtest.
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics.

    Example:
        result = client.public_symphony.backtest_symphony("sym-abc123")
        print(f"Sharpe: {result.stats.sharpe_ratio}")
    """
    payload = {
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self._client.post(
        f"/api/v1/public/symphonies/{symphony_id}/backtest",
        json=payload,
    )
    return BacktestResult.model_validate(raw_response)

backtest_symphony_v2(symphony_id, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a backtest for a public symphony by its ID (v2).

Parameters:

Name	Type	Description	Default
symphony_id	str	The ID of the symphony to backtest.	required
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics.

Example

result = client.public_symphony.backtest_symphony_v2("sym-abc123") print(f"Sharpe: {result.stats.sharpe_ratio}")

Source code in composer/resources/public_symphony.py

def backtest_symphony_v2(
    self,
    symphony_id: str,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """
    Run a backtest for a public symphony by its ID (v2).

    Args:
        symphony_id: The ID of the symphony to backtest.
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics.

    Example:
         result = client.public_symphony.backtest_symphony_v2("sym-abc123")
         print(f"Sharpe: {result.stats.sharpe_ratio}")
    """
    payload = {
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self._client.post(
        f"/api/v2/public/symphonies/{symphony_id}/backtest",
        json=payload,
    )
    return BacktestResult.model_validate(raw_response)

get_indicators()

Get a list of all available technical indicators.

Returns a list of indicators that can be used in symphony conditions, including moving averages, RSI, standard deviation, etc.

Returns

List[Indicator]: List of available technical indicators with their
    parameters and metadata.

Example

indicators = client.public_symphony.get_indicators() for indicator in indicators:

...     print(f"{indicator.name}: {indicator.description}")

Source code in composer/resources/public_symphony.py

def get_indicators(self) -> list[Indicator]:
    """
    Get a list of all available technical indicators.

    Returns a list of indicators that can be used in symphony conditions,
    including moving averages, RSI, standard deviation, etc.

    Returns
    -------
        List[Indicator]: List of available technical indicators with their
            parameters and metadata.

    Example:
         indicators = client.public_symphony.get_indicators()
         for indicator in indicators:
        ...     print(f"{indicator.name}: {indicator.description}")
    """
    response = self._client.get("/api/v1/public/symphony-scores/indicators")
    return [Indicator.model_validate(item) for item in response]

get_score(symphony_id, score_version='v1')

Get an existing symphony's EDN (score).

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required
score_version	str	Score version to retrieve ("v1" or "v2"). Defaults to "v1".	'v1'

Returns

SymphonyDefinition: The symphony score/parsed EDN structure.

Example

score = client.public_symphony.get_score("sym-abc123") print(score.name) print(score.rebalance)

Source code in composer/resources/public_symphony.py

def get_score(self, symphony_id: str, score_version: str = "v1") -> SymphonyDefinition:
    """
    Get an existing symphony's EDN (score).

    Args:
        symphony_id: Unique identifier for the symphony.
        score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

    Returns
    -------
        SymphonyDefinition: The symphony score/parsed EDN structure.

    Example:
         score = client.public_symphony.get_score("sym-abc123")
         print(score.name)
         print(score.rebalance)
    """
    params = {"score_version": score_version}
    response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/score", params=params)
    return SymphonyDefinition.model_validate(response)

get_symphony(symphony_id)

Get detailed information about a symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

SymphonyDetail: Detailed symphony information including stats,
    backtest data, and metadata.

Example

symphony = client.public_symphony.get_symphony("sym-abc123") print(f"Name: {symphony.name}") print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")

Source code in composer/resources/public_symphony.py

def get_symphony(self, symphony_id: str) -> SymphonyDetail:
    """
    Get detailed information about a symphony.

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        SymphonyDetail: Detailed symphony information including stats,
            backtest data, and metadata.

    Example:
         symphony = client.public_symphony.get_symphony("sym-abc123")
         print(f"Name: {symphony.name}")
         print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")
    """
    response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}")
    return SymphonyDetail.model_validate(response)

get_symphony_meta(symphony_ids)

Get metadata for symphonies by their IDs.

Parameters:

Name	Type	Description	Default
symphony_ids	list[str]	List of symphony IDs to look up.	required

Returns

List[SymphonyMeta]: List of symphony metadata objects.

Example

meta = client.public_symphony.get_symphony_meta(["sym-abc123"]) for symphony in meta:

...     print(f"{symphony.name}: {symphony.rebalance_frequency}")

Source code in composer/resources/public_symphony.py

def get_symphony_meta(self, symphony_ids: list[str]) -> list[SymphonyMeta]:
    """
    Get metadata for symphonies by their IDs.

    Args:
        symphony_ids: List of symphony IDs to look up.

    Returns
    -------
        List[SymphonyMeta]: List of symphony metadata objects.

    Example:
         meta = client.public_symphony.get_symphony_meta(["sym-abc123"])
         for symphony in meta:
        ...     print(f"{symphony.name}: {symphony.rebalance_frequency}")
    """
    response = self._client.post(
        "/api/v1/public/meta/symphonies", json={"symphony_id": symphony_ids}
    )
    result = SymphonyMetaResponse.model_validate(response)
    return result.symphonies

get_tickers(symphony_id)

Get all tickers used in a symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

List[str]: List of ticker symbols used in the symphony.

Example

tickers = client.public_symphony.get_tickers("sym-abc123") print(f"Symphony uses: {', '.join(tickers)}")

Source code in composer/resources/public_symphony.py

def get_tickers(self, symphony_id: str) -> list[str]:
    """
    Get all tickers used in a symphony.

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        List[str]: List of ticker symbols used in the symphony.

    Example:
         tickers = client.public_symphony.get_tickers("sym-abc123")
         print(f"Symphony uses: {', '.join(tickers)}")
    """
    response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/tickers")
    result = TickersResponse.model_validate(response)
    return result.tickers

get_version_score(symphony_id, version_id, score_version='v1')

Get an existing symphony version's EDN (score).

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required
version_id	str	Unique identifier for the symphony version.	required
score_version	str	Score version to retrieve ("v1" or "v2"). Defaults to "v1".	'v1'

Returns

SymphonyDefinition: The symphony score/parsed EDN structure.

Example

score = client.public_symphony.get_version_score(

...     "sym-abc123",
...     "ver-xyz789"
... )
 print(score.name)
 print(score.rebalance)

Source code in composer/resources/public_symphony.py

def get_version_score(
    self, symphony_id: str, version_id: str, score_version: str = "v1"
) -> SymphonyDefinition:
    """
    Get an existing symphony version's EDN (score).

    Args:
        symphony_id: Unique identifier for the symphony.
        version_id: Unique identifier for the symphony version.
        score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

    Returns
    -------
        SymphonyDefinition: The symphony score/parsed EDN structure.

    Example:
         score = client.public_symphony.get_version_score(
        ...     "sym-abc123",
        ...     "ver-xyz789"
        ... )
         print(score.name)
         print(score.rebalance)
    """
    params = {"score_version": score_version}
    response = self._client.get(
        f"/api/v1/public/symphonies/{symphony_id}/versions/{version_id}/score",
        params=params,
    )
    return SymphonyDefinition.model_validate(response)

get_versions(symphony_id)

Get all versions for a symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

List[SymphonyVersionInfo]: List of version information.

Example

versions = client.public_symphony.get_versions("sym-abc123") for version in versions:

...     print(f"Version {version.version_id}: {version.created_at}")

Source code in composer/resources/public_symphony.py

def get_versions(self, symphony_id: str) -> list[SymphonyVersionInfo]:
    """
    Get all versions for a symphony.

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        List[SymphonyVersionInfo]: List of version information.

    Example:
         versions = client.public_symphony.get_versions("sym-abc123")
         for version in versions:
        ...     print(f"Version {version.version_id}: {version.created_at}")
    """
    response = self._client.get(f"/api/v1/public/symphonies/{symphony_id}/versions")
    return [SymphonyVersionInfo.model_validate(item) for item in response]

User Symphony Resource

composer.resources.user_symphony.UserSymphony

Authenticated symphony endpoints.

These endpoints require authentication and provide access to the user's own symphonies as well as public symphonies with additional data.

Source code in composer/resources/user_symphony.py

class UserSymphony:
    """Authenticated symphony endpoints.

    These endpoints require authentication and provide access to the user's
    own symphonies as well as public symphonies with additional data.
    """

    def __init__(self, http_client: HTTPClient):
        self._client = http_client

    def get_indicators(self) -> list[Indicator]:
        """Get a list of all available technical indicators (authenticated).

        Same as the public endpoint but may include additional indicators
        or data for authenticated users.

        Returns
        -------
            List[Indicator]: List of available technical indicators with their
                parameters and metadata.

        Example:
             indicators = client.user_symphony.get_indicators()
             for indicator in indicators:
            ...     print(f"{indicator.name}: {indicator.description}")

        """
        response = self._client.get("/api/v1/symphony-scores/indicators")
        return [Indicator.model_validate(item) for item in response]

    def get_symphony(self, symphony_id: str) -> SymphonyDetail:
        """Get detailed information about a symphony (authenticated).

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            SymphonyDetail: Detailed symphony information including stats,
                backtest data, and metadata.

        Example:
             symphony = client.user_symphony.get_symphony("sym-abc123")
             print(f"Name: {symphony.name}")
             print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")

        """
        response = self._client.get(f"/api/v1/symphonies/{symphony_id}")
        return SymphonyDetail.model_validate(response)

    def get_versions(self, symphony_id: str) -> list[SymphonyVersionInfo]:
        """Get all versions for a symphony (authenticated).

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            List[SymphonyVersionInfo]: List of version information.

        Example:
             versions = client.user_symphony.get_versions("sym-abc123")
             for version in versions:
            ...     print(f"Version {version.version_id}: {version.created_at}")

        """
        response = self._client.get(f"/api/v1/symphonies/{symphony_id}/versions")
        return [SymphonyVersionInfo.model_validate(item) for item in response]

    def get_version_score(
        self, symphony_id: str, version_id: str, score_version: str = "v1"
    ) -> SymphonyDefinition:
        """Get an existing symphony version's EDN (score) - authenticated.

        Args:
            symphony_id: Unique identifier for the symphony.
            version_id: Unique identifier for the symphony version.
            score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

        Returns
        -------
            SymphonyDefinition: The symphony score/parsed EDN structure.

        Example:
             score = client.user_symphony.get_version_score(
            ...     "sym-abc123",
            ...     "ver-xyz789"
            ... )
             print(score.name)
             print(score.rebalance)

        """
        params = {"score_version": score_version}
        response = self._client.get(
            f"/api/v1/symphonies/{symphony_id}/versions/{version_id}/score",
            params=params,
        )
        return SymphonyDefinition.model_validate(response)

    def get_score(self, symphony_id: str, score_version: str = "v1") -> SymphonyDefinition:
        """Get an existing symphony's EDN (score) - authenticated.

        Args:
            symphony_id: Unique identifier for the symphony.
            score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

        Returns
        -------
            SymphonyDefinition: The symphony score/parsed EDN structure.

        Example:
             score = client.user_symphony.get_score("sym-abc123")
             print(score.name)
             print(score.rebalance)

        """
        params = {"score_version": score_version}
        response = self._client.get(f"/api/v1/symphonies/{symphony_id}/score", params=params)
        print(response)
        return SymphonyDefinition.model_validate(response)

    def get_score_extended(self, symphony_id: str) -> ScoreExtendedResponse:
        """Get a symphony's score along with suggested modifications.

        Args:
            symphony_id: Unique identifier for the symphony.

        Returns
        -------
            ScoreExtendedResponse: Contains the symphony score and suggested modifications.

        Example:
              result = client.user_symphony.get_score_extended("sym-abc123")
              print(result.score.name)
              for mod in result.modifications:
                  print(f"  Modification: {mod}")

        """
        response = self._client.get(f"/api/v1/symphonies/{symphony_id}/score-extended")

        modifications = []
        for mod in response.get("modifications", []):
            op = mod.get("op")
            if op == "FIND_AND_REPLACE":
                modifications.append(FindAndReplaceOperation.model_validate(mod))
            elif op == "COMPRESS_NESTED_IFS":
                modifications.append(CompressNestedIfsModification.model_validate(mod))
            else:
                modifications.append(mod)

        return ScoreExtendedResponse(
            score=SymphonyDefinition.model_validate(response.get("score")),
            modifications=modifications,
        )

    def from_defsymphony(self, code: str) -> SymphonyDefinition:
        """Convert defsymphony DSL text to symphony score.

        Args:
            code: The defsymphony DSL text to convert.

        Returns
        -------
            SymphonyDefinition: The parsed symphony score structure.

        Example:
             code = '(defsymphony "My Strategy" {:asset-class "EQUITIES"} ... )'
             score = client.user_symphony.from_defsymphony(code)
             print(score.name)
             print(score.rebalance)

        """
        request = DefSymphonyRequest(code=code)
        response = self._client.post(
            "/api/v1/symphony-scores/from-defsymphony",
            json=request.model_dump(),
        )
        return SymphonyDefinition.model_validate(response)

    def modify_symphony(
        self, symphony_id: str, old_ticker: str, new_ticker: str
    ) -> ModifySymphonyResponse:
        """Programmatically modify a symphony by finding and replacing a ticker.

        Args:
            symphony_id: Unique identifier for the symphony to modify.
            old_ticker: The ticker symbol to find (e.g., "SPY").
            new_ticker: The ticker symbol to replace with (e.g., "TQQQ").

        Returns
        -------
            ModifySymphonyResponse: Contains the symphony_id and version_id of
                the modified symphony.

        Example:
             result = client.user_symphony.modify_symphony(
            ...     "fk6VGRDAAgiH120TfUPS",
            ...     "SPY",
            ...     "TQQQ"
            ... )
             print(f"Modified symphony: {result.symphony_id}, version: {result.version_id}")

        """
        request_body = {
            "op": "FIND_AND_REPLACE",
            "old_ticker": old_ticker,
            "new_ticker": new_ticker,
        }
        response = self._client.post(
            f"/api/v1/symphonies/{symphony_id}/modify",
            json=request_body,
        )
        return ModifySymphonyResponse.model_validate(response)

    def update_symphony(
        self,
        symphony_id: str,
        name: str | None = None,
        description: str | None = None,
        color: str | None = None,
        hashtag: str | None = None,
        tags: list[str] | None = None,
        symphony: dict[str, Any] | None = None,
        benchmarks: list[dict[str, Any]] | None = None,
        share_with_everyone: bool | None = None,
    ) -> UpdateSymphonyResponse:
        """Update an existing symphony.

        Args:
            symphony_id: Unique identifier for the symphony to update.
            name: Optional new name for the symphony.
            description: Optional new description.
            color: Optional new color (hex code).
            hashtag: Optional new hashtag.
            tags: Optional list of tags.
            symphony: Optional symphony score/EDN definition.
            benchmarks: Optional list of benchmark configurations.
            share_with_everyone: Optional whether to share with everyone.

        Returns
        -------
            UpdateSymphonyResponse: Contains existing_version_id and version_id.

        Example:
             result = client.user_symphony.update_symphony(
            ...     "fk6VGRDAAgiH120TfUPS",
            ...     name="New Name",
            ...     description="New description"
            ... )
             print(f"New version: {result.version_id}")

        """
        request_body: dict[str, Any] = {}
        if name is not None:
            request_body["name"] = name
        if description is not None:
            request_body["description"] = description
        if color is not None:
            request_body["color"] = color
        if hashtag is not None:
            request_body["hashtag"] = hashtag
        if tags is not None:
            request_body["tags"] = tags
        if symphony is not None:
            request_body["symphony"] = symphony
        if benchmarks is not None:
            request_body["benchmarks"] = benchmarks
        if share_with_everyone is not None:
            request_body["share_with_everyone"] = share_with_everyone

        response = self._client.put(
            f"/api/v1/symphonies/{symphony_id}",
            json=request_body,
        )
        return UpdateSymphonyResponse.model_validate(response)

    def delete_symphony(self, symphony_id: str) -> None:
        """Delete an existing symphony.

        Args:
            symphony_id: Unique identifier for the symphony to delete.

        Example:
             client.user_symphony.delete_symphony("fk6VGRDAAgiH120TfUPS")

        """
        self._client.delete(f"/api/v1/symphonies/{symphony_id}")

    def submit_to_community(self, symphony_id: str) -> dict[str, Any]:
        """Submit a symphony to the Composer Community.

        Submitted symphonies will first be reviewed by an internal panel
        before being added to the Community search on Discover.

        Args:
            symphony_id: Unique identifier for the symphony to submit.

        Returns
        -------
            Dict[str, Any]: Response from the API.

        Example:
             result = client.user_symphony.submit_to_community("fk6VGRDAAgiH120TfUPS")
             print(result)

        """
        response = self._client.put(f"/api/v1/symphonies/{symphony_id}/submit-to-community")
        return response

    def create_symphony(
        self,
        name: str,
        asset_class: str = "EQUITIES",
        description: str | None = None,
        color: str | None = None,
        hashtag: str = "",
        tags: list[str] | None = None,
        symphony: SymphonyDefinition | dict[str, Any] | None = None,
        benchmarks: list[dict[str, Any]] | None = None,
    ) -> CreateSymphonyResponse:
        """Create a new symphony.

        Args:
            name: Name of the symphony.
            asset_class: Asset class (EQUITIES or CRYPTO). Defaults to EQUITIES.
            description: Optional description.
            color: Optional color (hex code). If not provided, a random color with
                good contrast for black text is generated.
            hashtag: Hashtag for the symphony (defaults to empty string).
            tags: Optional list of tags.
            symphony: Optional symphony definition (SymphonyDefinition model or dict).
            benchmarks: Optional list of benchmark configurations.

        Returns
        -------
            CreateSymphonyResponse: Contains symphony_id and version_id.

        Example:
             from composer.models.common import Asset, WeightCashEqual, SymphonyDefinition
             symphony = SymphonyDefinition(name="My Strategy", rebalance="daily", children=[...])
             result = client.user_symphony.create_symphony(
            ...     name="My Strategy",
            ...     symphony=symphony
            ... )
             print(f"Created: {result.symphony_id}")

        Example with dict:
             score_data = {"step": "root", "name": "My Strategy", ...}
             result = client.user_symphony.create_symphony(
            ...     name="My Strategy",
            ...     symphony=score_data
            ... )

        """
        request_body: dict[str, Any] = {"name": name, "asset_class": asset_class}
        if description is not None:
            request_body["description"] = description
        if color is None:
            color = _generate_random_color()
        request_body["color"] = color
        request_body["hashtag"] = hashtag
        if tags is not None:
            request_body["tags"] = tags
        if symphony is not None:
            if isinstance(symphony, dict):
                raw_value = symphony
            else:
                raw_value = symphony.model_dump(by_alias=True, mode="json", exclude_none=True)
            request_body["symphony"] = {"raw_value": raw_value}
        if benchmarks is not None:
            request_body["benchmarks"] = benchmarks

        response = self._client.post("/api/v1/symphonies", json=request_body)
        return CreateSymphonyResponse.model_validate(response)

    def copy_symphony(
        self,
        symphony_id: str,
        name: str | None = None,
        color: str | None = None,
        hashtag: str | None = None,
        benchmarks: list[dict[str, Any]] | None = None,
    ) -> CopySymphonyResponse:
        """Copy an existing symphony.

        Args:
            symphony_id: ID of the symphony to copy.
            name: Optional new name for the copied symphony.
            color: Optional color for the copied symphony.
            hashtag: Optional hashtag for the copied symphony.
            benchmarks: Optional benchmark configurations.

        Returns
        -------
            CopySymphonyResponse: Contains the new symphony_id and version_id.

        Example:
             result = client.user_symphony.copy_symphony(
            ...     "fk6VGRDAAgiH120TfUPS",
            ...     name="My Copy"
            ... )
             print(f"Copied: {result.symphony_id}")

        """
        request_body: dict[str, Any] = {}
        if name is not None:
            request_body["name"] = name
        if color is not None:
            request_body["color"] = color
        if hashtag is not None:
            request_body["hashtag"] = hashtag
        if benchmarks is not None:
            request_body["benchmarks"] = benchmarks

        response = self._client.post(
            f"/api/v1/symphonies/{symphony_id}/copy",
            json=request_body,
        )
        return CopySymphonyResponse.model_validate(response)

    def backtest_symphony(
        self,
        symphony_id: str,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """Run a backtest for an existing symphony by its ID.

        Args:
            symphony_id: The ID of the symphony to backtest.
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics.

        Example:
             result = client.user_symphony.backtest_symphony("fk6VGRDAAgiH120TfUPS")
             print(f"Sharpe: {result.stats.sharpe_ratio}")

        """
        payload = {
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self._client.post(
            f"/api/v1/symphonies/{symphony_id}/backtest",
            json=payload,
        )
        return BacktestResult.model_validate(raw_response)

    def backtest_symphony_v2(
        self,
        symphony_id: str,
        capital: float = 10000.0,
        abbreviate_days: int | None = None,
        apply_reg_fee: bool = True,
        apply_taf_fee: bool = True,
        apply_subscription: ApplySubscription = ApplySubscription.NONE,
        backtest_version: BacktestVersion = BacktestVersion.V2,
        slippage_percent: float = 0.0001,
        spread_markup: float = 0.0,
        start_date: str | None = None,
        end_date: str | None = None,
        broker: Broker = Broker.ALPACA_WHITE_LABEL,
        benchmark_symphonies: list[str] | None = None,
        benchmark_tickers: list[str] | None = None,
        sparkgraph_color: str | None = None,
    ) -> BacktestResult:
        """Run a backtest for an existing symphony by its ID (v2).

        Args:
            symphony_id: The ID of the symphony to backtest.
            capital: Initial capital for the backtest (default: 10000.0).
            abbreviate_days: Number of days to abbreviate the backtest (for testing).
            apply_reg_fee: Whether to apply regulatory fees (SEC fees).
            apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
            apply_subscription: Composer subscription level to simulate (affects fees).
            backtest_version: Backtest engine version to use.
            slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
            spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
            start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
            end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
            broker: Broker to simulate for fee calculations.
            benchmark_symphonies: List of symphony IDs to use as benchmarks.
            benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
            sparkgraph_color: Custom color for performance chart.

        Returns
        -------
            BacktestResult: Parsed backtest result with all statistics.

        Example:
             result = client.user_symphony.backtest_symphony_v2("fk6VGRDAAgiH120TfUPS")
             print(f"Sharpe: {result.stats.sharpe_ratio}")

        """
        payload = {
            "capital": capital,
            "abbreviate_days": abbreviate_days,
            "apply_reg_fee": apply_reg_fee,
            "apply_taf_fee": apply_taf_fee,
            "apply_subscription": apply_subscription.value if apply_subscription else None,
            "backtest_version": backtest_version.value if backtest_version else None,
            "slippage_percent": slippage_percent,
            "spread_markup": spread_markup,
            "start_date": start_date,
            "end_date": end_date,
            "broker": broker.value if broker else None,
            "benchmark_symphonies": benchmark_symphonies,
            "benchmark_tickers": benchmark_tickers,
            "sparkgraph_color": sparkgraph_color,
        }
        payload = {k: v for k, v in payload.items() if v is not None}

        raw_response = self._client.post(
            f"/api/v2/symphonies/{symphony_id}/backtest",
            json=payload,
        )
        return BacktestResult.model_validate(raw_response)

    def update_symphony_nodes(
        self,
        symphony_id: str,
        version_id: str,
        updates: list[dict[str, Any]],
    ) -> UpdateSymphonyNodesResponse:
        """Update nodes of a symphony version's EDN.

        Args:
            symphony_id: Unique identifier for the symphony.
            version_id: Unique identifier for the symphony version (must be latest).
            updates: List of node updates to apply.

        Returns
        -------
            UpdateSymphonyNodesResponse: Response containing symphony_id and version_id.

        Example:
             result = client.user_symphony.update_symphony_nodes(
            ...     "fk6VGRDAAgiH120TfUPS",
            ...     "v1",
            ...     [{"id": "node-123", "ticker": "QQQ"}]
            ... )
             print(result.symphony_id, result.version_id)

        """
        request_body = {"updates": updates}
        response = self._client.patch(
            f"/api/v1/symphonies/{symphony_id}/versions/{version_id}/score/nodes",
            json=request_body,
        )
        return UpdateSymphonyNodesResponse.model_validate(response)

backtest_symphony(symphony_id, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a backtest for an existing symphony by its ID.

Parameters:

Name	Type	Description	Default
symphony_id	str	The ID of the symphony to backtest.	required
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics.

Example

result = client.user_symphony.backtest_symphony("fk6VGRDAAgiH120TfUPS") print(f"Sharpe: {result.stats.sharpe_ratio}")

Source code in composer/resources/user_symphony.py

def backtest_symphony(
    self,
    symphony_id: str,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """Run a backtest for an existing symphony by its ID.

    Args:
        symphony_id: The ID of the symphony to backtest.
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics.

    Example:
         result = client.user_symphony.backtest_symphony("fk6VGRDAAgiH120TfUPS")
         print(f"Sharpe: {result.stats.sharpe_ratio}")

    """
    payload = {
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self._client.post(
        f"/api/v1/symphonies/{symphony_id}/backtest",
        json=payload,
    )
    return BacktestResult.model_validate(raw_response)

backtest_symphony_v2(symphony_id, capital=10000.0, abbreviate_days=None, apply_reg_fee=True, apply_taf_fee=True, apply_subscription=ApplySubscription.NONE, backtest_version=BacktestVersion.V2, slippage_percent=0.0001, spread_markup=0.0, start_date=None, end_date=None, broker=Broker.ALPACA_WHITE_LABEL, benchmark_symphonies=None, benchmark_tickers=None, sparkgraph_color=None)

Run a backtest for an existing symphony by its ID (v2).

Parameters:

Name	Type	Description	Default
symphony_id	str	The ID of the symphony to backtest.	required
capital	float	Initial capital for the backtest (default: 10000.0).	10000.0
abbreviate_days	int | None	Number of days to abbreviate the backtest (for testing).	None
apply_reg_fee	bool	Whether to apply regulatory fees (SEC fees).	True
apply_taf_fee	bool	Whether to apply TAF (Trading Activity Fee).	True
apply_subscription	ApplySubscription	Composer subscription level to simulate (affects fees).	NONE
backtest_version	BacktestVersion	Backtest engine version to use.	V2
slippage_percent	float	Slippage assumption as decimal (0.0001 = 0.01%).	0.0001
spread_markup	float	Bid-ask spread markup as decimal (0.001 = 0.1%).	0.0
start_date	str | None	Backtest start date (YYYY-MM-DD). Defaults to earliest available data.	None
end_date	str | None	Backtest end date (YYYY-MM-DD). Defaults to latest available data.	None
broker	Broker	Broker to simulate for fee calculations.	ALPACA_WHITE_LABEL
benchmark_symphonies	list[str] | None	List of symphony IDs to use as benchmarks.	None
benchmark_tickers	list[str] | None	List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).	None
sparkgraph_color	str | None	Custom color for performance chart.	None

Returns

BacktestResult: Parsed backtest result with all statistics.

Example

result = client.user_symphony.backtest_symphony_v2("fk6VGRDAAgiH120TfUPS") print(f"Sharpe: {result.stats.sharpe_ratio}")

Source code in composer/resources/user_symphony.py

def backtest_symphony_v2(
    self,
    symphony_id: str,
    capital: float = 10000.0,
    abbreviate_days: int | None = None,
    apply_reg_fee: bool = True,
    apply_taf_fee: bool = True,
    apply_subscription: ApplySubscription = ApplySubscription.NONE,
    backtest_version: BacktestVersion = BacktestVersion.V2,
    slippage_percent: float = 0.0001,
    spread_markup: float = 0.0,
    start_date: str | None = None,
    end_date: str | None = None,
    broker: Broker = Broker.ALPACA_WHITE_LABEL,
    benchmark_symphonies: list[str] | None = None,
    benchmark_tickers: list[str] | None = None,
    sparkgraph_color: str | None = None,
) -> BacktestResult:
    """Run a backtest for an existing symphony by its ID (v2).

    Args:
        symphony_id: The ID of the symphony to backtest.
        capital: Initial capital for the backtest (default: 10000.0).
        abbreviate_days: Number of days to abbreviate the backtest (for testing).
        apply_reg_fee: Whether to apply regulatory fees (SEC fees).
        apply_taf_fee: Whether to apply TAF (Trading Activity Fee).
        apply_subscription: Composer subscription level to simulate (affects fees).
        backtest_version: Backtest engine version to use.
        slippage_percent: Slippage assumption as decimal (0.0001 = 0.01%).
        spread_markup: Bid-ask spread markup as decimal (0.001 = 0.1%).
        start_date: Backtest start date (YYYY-MM-DD). Defaults to earliest available data.
        end_date: Backtest end date (YYYY-MM-DD). Defaults to latest available data.
        broker: Broker to simulate for fee calculations.
        benchmark_symphonies: List of symphony IDs to use as benchmarks.
        benchmark_tickers: List of ticker symbols to use as benchmarks (e.g., ['SPY', 'QQQ']).
        sparkgraph_color: Custom color for performance chart.

    Returns
    -------
        BacktestResult: Parsed backtest result with all statistics.

    Example:
         result = client.user_symphony.backtest_symphony_v2("fk6VGRDAAgiH120TfUPS")
         print(f"Sharpe: {result.stats.sharpe_ratio}")

    """
    payload = {
        "capital": capital,
        "abbreviate_days": abbreviate_days,
        "apply_reg_fee": apply_reg_fee,
        "apply_taf_fee": apply_taf_fee,
        "apply_subscription": apply_subscription.value if apply_subscription else None,
        "backtest_version": backtest_version.value if backtest_version else None,
        "slippage_percent": slippage_percent,
        "spread_markup": spread_markup,
        "start_date": start_date,
        "end_date": end_date,
        "broker": broker.value if broker else None,
        "benchmark_symphonies": benchmark_symphonies,
        "benchmark_tickers": benchmark_tickers,
        "sparkgraph_color": sparkgraph_color,
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    raw_response = self._client.post(
        f"/api/v2/symphonies/{symphony_id}/backtest",
        json=payload,
    )
    return BacktestResult.model_validate(raw_response)

copy_symphony(symphony_id, name=None, color=None, hashtag=None, benchmarks=None)

Copy an existing symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	ID of the symphony to copy.	required
name	str | None	Optional new name for the copied symphony.	None
color	str | None	Optional color for the copied symphony.	None
hashtag	str | None	Optional hashtag for the copied symphony.	None
benchmarks	list[dict[str, Any]] | None	Optional benchmark configurations.	None

Returns

CopySymphonyResponse: Contains the new symphony_id and version_id.

Example

result = client.user_symphony.copy_symphony(

...     "fk6VGRDAAgiH120TfUPS",
...     name="My Copy"
... )
 print(f"Copied: {result.symphony_id}")

Source code in composer/resources/user_symphony.py

def copy_symphony(
    self,
    symphony_id: str,
    name: str | None = None,
    color: str | None = None,
    hashtag: str | None = None,
    benchmarks: list[dict[str, Any]] | None = None,
) -> CopySymphonyResponse:
    """Copy an existing symphony.

    Args:
        symphony_id: ID of the symphony to copy.
        name: Optional new name for the copied symphony.
        color: Optional color for the copied symphony.
        hashtag: Optional hashtag for the copied symphony.
        benchmarks: Optional benchmark configurations.

    Returns
    -------
        CopySymphonyResponse: Contains the new symphony_id and version_id.

    Example:
         result = client.user_symphony.copy_symphony(
        ...     "fk6VGRDAAgiH120TfUPS",
        ...     name="My Copy"
        ... )
         print(f"Copied: {result.symphony_id}")

    """
    request_body: dict[str, Any] = {}
    if name is not None:
        request_body["name"] = name
    if color is not None:
        request_body["color"] = color
    if hashtag is not None:
        request_body["hashtag"] = hashtag
    if benchmarks is not None:
        request_body["benchmarks"] = benchmarks

    response = self._client.post(
        f"/api/v1/symphonies/{symphony_id}/copy",
        json=request_body,
    )
    return CopySymphonyResponse.model_validate(response)

create_symphony(name, asset_class='EQUITIES', description=None, color=None, hashtag='', tags=None, symphony=None, benchmarks=None)

Create a new symphony.

Parameters:

Name	Type	Description	Default
name	str	Name of the symphony.	required
asset_class	str	Asset class (EQUITIES or CRYPTO). Defaults to EQUITIES.	'EQUITIES'
description	str | None	Optional description.	None
color	str | None	Optional color (hex code). If not provided, a random color with good contrast for black text is generated.	None
hashtag	str	Hashtag for the symphony (defaults to empty string).	''
tags	list[str] | None	Optional list of tags.	None
symphony	SymphonyDefinition | dict[str, Any] | None	Optional symphony definition (SymphonyDefinition model or dict).	None
benchmarks	list[dict[str, Any]] | None	Optional list of benchmark configurations.	None

Returns

CreateSymphonyResponse: Contains symphony_id and version_id.

Example

from composer.models.common import Asset, WeightCashEqual, SymphonyDefinition symphony = SymphonyDefinition(name="My Strategy", rebalance="daily", children=[...]) result = client.user_symphony.create_symphony(

...     name="My Strategy",
...     symphony=symphony
... )
 print(f"Created: {result.symphony_id}")

Example with dict

score_data = {"step": "root", "name": "My Strategy", ...} result = client.user_symphony.create_symphony(

...     name="My Strategy",
...     symphony=score_data
... )

Source code in composer/resources/user_symphony.py

def create_symphony(
    self,
    name: str,
    asset_class: str = "EQUITIES",
    description: str | None = None,
    color: str | None = None,
    hashtag: str = "",
    tags: list[str] | None = None,
    symphony: SymphonyDefinition | dict[str, Any] | None = None,
    benchmarks: list[dict[str, Any]] | None = None,
) -> CreateSymphonyResponse:
    """Create a new symphony.

    Args:
        name: Name of the symphony.
        asset_class: Asset class (EQUITIES or CRYPTO). Defaults to EQUITIES.
        description: Optional description.
        color: Optional color (hex code). If not provided, a random color with
            good contrast for black text is generated.
        hashtag: Hashtag for the symphony (defaults to empty string).
        tags: Optional list of tags.
        symphony: Optional symphony definition (SymphonyDefinition model or dict).
        benchmarks: Optional list of benchmark configurations.

    Returns
    -------
        CreateSymphonyResponse: Contains symphony_id and version_id.

    Example:
         from composer.models.common import Asset, WeightCashEqual, SymphonyDefinition
         symphony = SymphonyDefinition(name="My Strategy", rebalance="daily", children=[...])
         result = client.user_symphony.create_symphony(
        ...     name="My Strategy",
        ...     symphony=symphony
        ... )
         print(f"Created: {result.symphony_id}")

    Example with dict:
         score_data = {"step": "root", "name": "My Strategy", ...}
         result = client.user_symphony.create_symphony(
        ...     name="My Strategy",
        ...     symphony=score_data
        ... )

    """
    request_body: dict[str, Any] = {"name": name, "asset_class": asset_class}
    if description is not None:
        request_body["description"] = description
    if color is None:
        color = _generate_random_color()
    request_body["color"] = color
    request_body["hashtag"] = hashtag
    if tags is not None:
        request_body["tags"] = tags
    if symphony is not None:
        if isinstance(symphony, dict):
            raw_value = symphony
        else:
            raw_value = symphony.model_dump(by_alias=True, mode="json", exclude_none=True)
        request_body["symphony"] = {"raw_value": raw_value}
    if benchmarks is not None:
        request_body["benchmarks"] = benchmarks

    response = self._client.post("/api/v1/symphonies", json=request_body)
    return CreateSymphonyResponse.model_validate(response)

delete_symphony(symphony_id)

Delete an existing symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony to delete.	required

Example

client.user_symphony.delete_symphony("fk6VGRDAAgiH120TfUPS")

Source code in composer/resources/user_symphony.py

def delete_symphony(self, symphony_id: str) -> None:
    """Delete an existing symphony.

    Args:
        symphony_id: Unique identifier for the symphony to delete.

    Example:
         client.user_symphony.delete_symphony("fk6VGRDAAgiH120TfUPS")

    """
    self._client.delete(f"/api/v1/symphonies/{symphony_id}")

from_defsymphony(code)

Convert defsymphony DSL text to symphony score.

Parameters:

Name	Type	Description	Default
code	str	The defsymphony DSL text to convert.	required

Returns

SymphonyDefinition: The parsed symphony score structure.

Example

code = '(defsymphony "My Strategy" {:asset-class "EQUITIES"} ... )' score = client.user_symphony.from_defsymphony(code) print(score.name) print(score.rebalance)

Source code in composer/resources/user_symphony.py

def from_defsymphony(self, code: str) -> SymphonyDefinition:
    """Convert defsymphony DSL text to symphony score.

    Args:
        code: The defsymphony DSL text to convert.

    Returns
    -------
        SymphonyDefinition: The parsed symphony score structure.

    Example:
         code = '(defsymphony "My Strategy" {:asset-class "EQUITIES"} ... )'
         score = client.user_symphony.from_defsymphony(code)
         print(score.name)
         print(score.rebalance)

    """
    request = DefSymphonyRequest(code=code)
    response = self._client.post(
        "/api/v1/symphony-scores/from-defsymphony",
        json=request.model_dump(),
    )
    return SymphonyDefinition.model_validate(response)

get_indicators()

Get a list of all available technical indicators (authenticated).

Same as the public endpoint but may include additional indicators or data for authenticated users.

Returns

List[Indicator]: List of available technical indicators with their
    parameters and metadata.

Example

indicators = client.user_symphony.get_indicators() for indicator in indicators:

...     print(f"{indicator.name}: {indicator.description}")

Source code in composer/resources/user_symphony.py

def get_indicators(self) -> list[Indicator]:
    """Get a list of all available technical indicators (authenticated).

    Same as the public endpoint but may include additional indicators
    or data for authenticated users.

    Returns
    -------
        List[Indicator]: List of available technical indicators with their
            parameters and metadata.

    Example:
         indicators = client.user_symphony.get_indicators()
         for indicator in indicators:
        ...     print(f"{indicator.name}: {indicator.description}")

    """
    response = self._client.get("/api/v1/symphony-scores/indicators")
    return [Indicator.model_validate(item) for item in response]

get_score(symphony_id, score_version='v1')

Get an existing symphony's EDN (score) - authenticated.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required
score_version	str	Score version to retrieve ("v1" or "v2"). Defaults to "v1".	'v1'

Returns

SymphonyDefinition: The symphony score/parsed EDN structure.

Example

score = client.user_symphony.get_score("sym-abc123") print(score.name) print(score.rebalance)

Source code in composer/resources/user_symphony.py

def get_score(self, symphony_id: str, score_version: str = "v1") -> SymphonyDefinition:
    """Get an existing symphony's EDN (score) - authenticated.

    Args:
        symphony_id: Unique identifier for the symphony.
        score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

    Returns
    -------
        SymphonyDefinition: The symphony score/parsed EDN structure.

    Example:
         score = client.user_symphony.get_score("sym-abc123")
         print(score.name)
         print(score.rebalance)

    """
    params = {"score_version": score_version}
    response = self._client.get(f"/api/v1/symphonies/{symphony_id}/score", params=params)
    print(response)
    return SymphonyDefinition.model_validate(response)

get_score_extended(symphony_id)

Get a symphony's score along with suggested modifications.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

ScoreExtendedResponse: Contains the symphony score and suggested modifications.

Example

result = client.user_symphony.get_score_extended("sym-abc123") print(result.score.name) for mod in result.modifications: print(f" Modification: {mod}")

Source code in composer/resources/user_symphony.py

def get_score_extended(self, symphony_id: str) -> ScoreExtendedResponse:
    """Get a symphony's score along with suggested modifications.

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        ScoreExtendedResponse: Contains the symphony score and suggested modifications.

    Example:
          result = client.user_symphony.get_score_extended("sym-abc123")
          print(result.score.name)
          for mod in result.modifications:
              print(f"  Modification: {mod}")

    """
    response = self._client.get(f"/api/v1/symphonies/{symphony_id}/score-extended")

    modifications = []
    for mod in response.get("modifications", []):
        op = mod.get("op")
        if op == "FIND_AND_REPLACE":
            modifications.append(FindAndReplaceOperation.model_validate(mod))
        elif op == "COMPRESS_NESTED_IFS":
            modifications.append(CompressNestedIfsModification.model_validate(mod))
        else:
            modifications.append(mod)

    return ScoreExtendedResponse(
        score=SymphonyDefinition.model_validate(response.get("score")),
        modifications=modifications,
    )

get_symphony(symphony_id)

Get detailed information about a symphony (authenticated).

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

SymphonyDetail: Detailed symphony information including stats,
    backtest data, and metadata.

Example

symphony = client.user_symphony.get_symphony("sym-abc123") print(f"Name: {symphony.name}") print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")

Source code in composer/resources/user_symphony.py

def get_symphony(self, symphony_id: str) -> SymphonyDetail:
    """Get detailed information about a symphony (authenticated).

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        SymphonyDetail: Detailed symphony information including stats,
            backtest data, and metadata.

    Example:
         symphony = client.user_symphony.get_symphony("sym-abc123")
         print(f"Name: {symphony.name}")
         print(f"Sharpe: {symphony.stats_oos_sharpe_ratio}")

    """
    response = self._client.get(f"/api/v1/symphonies/{symphony_id}")
    return SymphonyDetail.model_validate(response)

get_version_score(symphony_id, version_id, score_version='v1')

Get an existing symphony version's EDN (score) - authenticated.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required
version_id	str	Unique identifier for the symphony version.	required
score_version	str	Score version to retrieve ("v1" or "v2"). Defaults to "v1".	'v1'

Returns

SymphonyDefinition: The symphony score/parsed EDN structure.

Example

score = client.user_symphony.get_version_score(

...     "sym-abc123",
...     "ver-xyz789"
... )
 print(score.name)
 print(score.rebalance)

Source code in composer/resources/user_symphony.py

def get_version_score(
    self, symphony_id: str, version_id: str, score_version: str = "v1"
) -> SymphonyDefinition:
    """Get an existing symphony version's EDN (score) - authenticated.

    Args:
        symphony_id: Unique identifier for the symphony.
        version_id: Unique identifier for the symphony version.
        score_version: Score version to retrieve ("v1" or "v2"). Defaults to "v1".

    Returns
    -------
        SymphonyDefinition: The symphony score/parsed EDN structure.

    Example:
         score = client.user_symphony.get_version_score(
        ...     "sym-abc123",
        ...     "ver-xyz789"
        ... )
         print(score.name)
         print(score.rebalance)

    """
    params = {"score_version": score_version}
    response = self._client.get(
        f"/api/v1/symphonies/{symphony_id}/versions/{version_id}/score",
        params=params,
    )
    return SymphonyDefinition.model_validate(response)

get_versions(symphony_id)

Get all versions for a symphony (authenticated).

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required

Returns

List[SymphonyVersionInfo]: List of version information.

Example

versions = client.user_symphony.get_versions("sym-abc123") for version in versions:

...     print(f"Version {version.version_id}: {version.created_at}")

Source code in composer/resources/user_symphony.py

def get_versions(self, symphony_id: str) -> list[SymphonyVersionInfo]:
    """Get all versions for a symphony (authenticated).

    Args:
        symphony_id: Unique identifier for the symphony.

    Returns
    -------
        List[SymphonyVersionInfo]: List of version information.

    Example:
         versions = client.user_symphony.get_versions("sym-abc123")
         for version in versions:
        ...     print(f"Version {version.version_id}: {version.created_at}")

    """
    response = self._client.get(f"/api/v1/symphonies/{symphony_id}/versions")
    return [SymphonyVersionInfo.model_validate(item) for item in response]

modify_symphony(symphony_id, old_ticker, new_ticker)

Programmatically modify a symphony by finding and replacing a ticker.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony to modify.	required
old_ticker	str	The ticker symbol to find (e.g., "SPY").	required
new_ticker	str	The ticker symbol to replace with (e.g., "TQQQ").	required

Returns

ModifySymphonyResponse: Contains the symphony_id and version_id of
    the modified symphony.

Example

result = client.user_symphony.modify_symphony(

...     "fk6VGRDAAgiH120TfUPS",
...     "SPY",
...     "TQQQ"
... )
 print(f"Modified symphony: {result.symphony_id}, version: {result.version_id}")

Source code in composer/resources/user_symphony.py

def modify_symphony(
    self, symphony_id: str, old_ticker: str, new_ticker: str
) -> ModifySymphonyResponse:
    """Programmatically modify a symphony by finding and replacing a ticker.

    Args:
        symphony_id: Unique identifier for the symphony to modify.
        old_ticker: The ticker symbol to find (e.g., "SPY").
        new_ticker: The ticker symbol to replace with (e.g., "TQQQ").

    Returns
    -------
        ModifySymphonyResponse: Contains the symphony_id and version_id of
            the modified symphony.

    Example:
         result = client.user_symphony.modify_symphony(
        ...     "fk6VGRDAAgiH120TfUPS",
        ...     "SPY",
        ...     "TQQQ"
        ... )
         print(f"Modified symphony: {result.symphony_id}, version: {result.version_id}")

    """
    request_body = {
        "op": "FIND_AND_REPLACE",
        "old_ticker": old_ticker,
        "new_ticker": new_ticker,
    }
    response = self._client.post(
        f"/api/v1/symphonies/{symphony_id}/modify",
        json=request_body,
    )
    return ModifySymphonyResponse.model_validate(response)

submit_to_community(symphony_id)

Submit a symphony to the Composer Community.

Submitted symphonies will first be reviewed by an internal panel before being added to the Community search on Discover.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony to submit.	required

Returns

Dict[str, Any]: Response from the API.

Example

result = client.user_symphony.submit_to_community("fk6VGRDAAgiH120TfUPS") print(result)

Source code in composer/resources/user_symphony.py

def submit_to_community(self, symphony_id: str) -> dict[str, Any]:
    """Submit a symphony to the Composer Community.

    Submitted symphonies will first be reviewed by an internal panel
    before being added to the Community search on Discover.

    Args:
        symphony_id: Unique identifier for the symphony to submit.

    Returns
    -------
        Dict[str, Any]: Response from the API.

    Example:
         result = client.user_symphony.submit_to_community("fk6VGRDAAgiH120TfUPS")
         print(result)

    """
    response = self._client.put(f"/api/v1/symphonies/{symphony_id}/submit-to-community")
    return response

update_symphony(symphony_id, name=None, description=None, color=None, hashtag=None, tags=None, symphony=None, benchmarks=None, share_with_everyone=None)

Update an existing symphony.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony to update.	required
name	str | None	Optional new name for the symphony.	None
description	str | None	Optional new description.	None
color	str | None	Optional new color (hex code).	None
hashtag	str | None	Optional new hashtag.	None
tags	list[str] | None	Optional list of tags.	None
symphony	dict[str, Any] | None	Optional symphony score/EDN definition.	None
benchmarks	list[dict[str, Any]] | None	Optional list of benchmark configurations.	None
share_with_everyone	bool | None	Optional whether to share with everyone.	None

Returns

UpdateSymphonyResponse: Contains existing_version_id and version_id.

Example

result = client.user_symphony.update_symphony(

...     "fk6VGRDAAgiH120TfUPS",
...     name="New Name",
...     description="New description"
... )
 print(f"New version: {result.version_id}")

Source code in composer/resources/user_symphony.py

def update_symphony(
    self,
    symphony_id: str,
    name: str | None = None,
    description: str | None = None,
    color: str | None = None,
    hashtag: str | None = None,
    tags: list[str] | None = None,
    symphony: dict[str, Any] | None = None,
    benchmarks: list[dict[str, Any]] | None = None,
    share_with_everyone: bool | None = None,
) -> UpdateSymphonyResponse:
    """Update an existing symphony.

    Args:
        symphony_id: Unique identifier for the symphony to update.
        name: Optional new name for the symphony.
        description: Optional new description.
        color: Optional new color (hex code).
        hashtag: Optional new hashtag.
        tags: Optional list of tags.
        symphony: Optional symphony score/EDN definition.
        benchmarks: Optional list of benchmark configurations.
        share_with_everyone: Optional whether to share with everyone.

    Returns
    -------
        UpdateSymphonyResponse: Contains existing_version_id and version_id.

    Example:
         result = client.user_symphony.update_symphony(
        ...     "fk6VGRDAAgiH120TfUPS",
        ...     name="New Name",
        ...     description="New description"
        ... )
         print(f"New version: {result.version_id}")

    """
    request_body: dict[str, Any] = {}
    if name is not None:
        request_body["name"] = name
    if description is not None:
        request_body["description"] = description
    if color is not None:
        request_body["color"] = color
    if hashtag is not None:
        request_body["hashtag"] = hashtag
    if tags is not None:
        request_body["tags"] = tags
    if symphony is not None:
        request_body["symphony"] = symphony
    if benchmarks is not None:
        request_body["benchmarks"] = benchmarks
    if share_with_everyone is not None:
        request_body["share_with_everyone"] = share_with_everyone

    response = self._client.put(
        f"/api/v1/symphonies/{symphony_id}",
        json=request_body,
    )
    return UpdateSymphonyResponse.model_validate(response)

update_symphony_nodes(symphony_id, version_id, updates)

Update nodes of a symphony version's EDN.

Parameters:

Name	Type	Description	Default
symphony_id	str	Unique identifier for the symphony.	required
version_id	str	Unique identifier for the symphony version (must be latest).	required
updates	list[dict[str, Any]]	List of node updates to apply.	required

Returns

UpdateSymphonyNodesResponse: Response containing symphony_id and version_id.

Example

result = client.user_symphony.update_symphony_nodes(

...     "fk6VGRDAAgiH120TfUPS",
...     "v1",
...     [{"id": "node-123", "ticker": "QQQ"}]
... )
 print(result.symphony_id, result.version_id)

Source code in composer/resources/user_symphony.py

def update_symphony_nodes(
    self,
    symphony_id: str,
    version_id: str,
    updates: list[dict[str, Any]],
) -> UpdateSymphonyNodesResponse:
    """Update nodes of a symphony version's EDN.

    Args:
        symphony_id: Unique identifier for the symphony.
        version_id: Unique identifier for the symphony version (must be latest).
        updates: List of node updates to apply.

    Returns
    -------
        UpdateSymphonyNodesResponse: Response containing symphony_id and version_id.

    Example:
         result = client.user_symphony.update_symphony_nodes(
        ...     "fk6VGRDAAgiH120TfUPS",
        ...     "v1",
        ...     [{"id": "node-123", "ticker": "QQQ"}]
        ... )
         print(result.symphony_id, result.version_id)

    """
    request_body = {"updates": updates}
    response = self._client.patch(
        f"/api/v1/symphonies/{symphony_id}/versions/{version_id}/score/nodes",
        json=request_body,
    )
    return UpdateSymphonyNodesResponse.model_validate(response)

User Symphonies Resource

composer.resources.user_symphonies.UserSymphonies

User's symphony list endpoints.

These endpoints provide access to the user's library of symphonies, including both published and draft symphonies.

Source code in composer/resources/user_symphonies.py

class UserSymphonies:
    """User's symphony list endpoints.

    These endpoints provide access to the user's library of symphonies,
    including both published and draft symphonies.
    """

    def __init__(self, http_client: HTTPClient):
        self._client = http_client

    def list_symphonies(self) -> list[UserSymphony]:
        """Get all symphonies in the user's library.

        Returns a list of all symphonies created by or copied to the
        authenticated user's account.

        Returns
        -------
            List[UserSymphony]: List of user's symphonies with statistics.

        Example:
             symphonies = client.user_symphonies.list_symphonies()
             for symphony in symphonies:
            ...     print(f"{symphony.name}: Sharpe={symphony.oos_sharpe_ratio}")

        """
        response = self._client.get("/api/v1/user/symphonies")
        result = UserSymphoniesResponse.model_validate(response)
        return result.symphonies

    def list_drafts(self) -> list[DraftSymphony]:
        """Get all draft symphonies in the user's library.

        Returns a list of draft (unpublished) symphonies created by the
        authenticated user.

        Returns
        -------
            List[DraftSymphony]: List of user's draft symphonies with statistics.

        Example:
             drafts = client.user_symphonies.list_drafts()
             for draft in drafts:
            ...     print(f"Draft: {draft.name}")

        """
        response = self._client.get("/api/v1/user/symphonies/drafts")
        result = DraftSymphoniesResponse.model_validate(response)
        return result.symphonies

    def bulk_modify_symphonies(
        self, old_ticker: str, new_ticker: str, user_id: str | None = None
    ) -> list[dict[str, Any]]:
        """Bulk modify all symphonies in the user's library by finding and replacing a ticker.

        Args:
            old_ticker: The ticker symbol to find (e.g., "SPY").
            new_ticker: The ticker symbol to replace with (e.g., "TQQQ").
            user_id: Optional user ID. If not provided, uses the authenticated user.

        Returns
        -------
            List[Dict[str, Any]]: Response from the bulk modify operation.

        Example:
             result = client.user_symphonies.bulk_modify_symphonies(
            ...     "SPY",
            ...     "TQQQ"
            ... )
             print(result)

        """
        request_body = {
            "op": "FIND_AND_REPLACE",
            "old_ticker": old_ticker,
            "new_ticker": new_ticker,
        }
        if user_id:
            request_body["user_id"] = user_id
        response = self._client.post(
            "/api/v1/user/symphonies/modify",
            json=request_body,
        )
        return response

    def pubsub_modify_symphonies(
        self,
        subscription: str,
        message: dict[str, Any],
    ) -> dict[str, Any]:
        """Programmatically modify all of a user's symphonies via pub/sub.

        Args:
            subscription: The pub/sub subscription name.
            message: The message containing modification instructions.

        Returns
        -------
            Dict[str, Any]: Response from the API.

        Example:
             result = client.user_symphonies.pubsub_modify_symphonies(
            ...     subscription="my-subscription",
            ...     message={"publish_time": "2024-01-01", "data": {...}}
            ... )
             print(result)

        """
        request_body = {
            "subscription": subscription,
            "message": message,
        }
        response = self._client.post(
            "/api/v1/pubsub/symphonies/modify",
            json=request_body,
        )
        return response

bulk_modify_symphonies(old_ticker, new_ticker, user_id=None)

Bulk modify all symphonies in the user's library by finding and replacing a ticker.

Parameters:

Name	Type	Description	Default
old_ticker	str	The ticker symbol to find (e.g., "SPY").	required
new_ticker	str	The ticker symbol to replace with (e.g., "TQQQ").	required
user_id	str | None	Optional user ID. If not provided, uses the authenticated user.	None

Returns

List[Dict[str, Any]]: Response from the bulk modify operation.

Example

result = client.user_symphonies.bulk_modify_symphonies(

...     "SPY",
...     "TQQQ"
... )
 print(result)

Source code in composer/resources/user_symphonies.py

def bulk_modify_symphonies(
    self, old_ticker: str, new_ticker: str, user_id: str | None = None
) -> list[dict[str, Any]]:
    """Bulk modify all symphonies in the user's library by finding and replacing a ticker.

    Args:
        old_ticker: The ticker symbol to find (e.g., "SPY").
        new_ticker: The ticker symbol to replace with (e.g., "TQQQ").
        user_id: Optional user ID. If not provided, uses the authenticated user.

    Returns
    -------
        List[Dict[str, Any]]: Response from the bulk modify operation.

    Example:
         result = client.user_symphonies.bulk_modify_symphonies(
        ...     "SPY",
        ...     "TQQQ"
        ... )
         print(result)

    """
    request_body = {
        "op": "FIND_AND_REPLACE",
        "old_ticker": old_ticker,
        "new_ticker": new_ticker,
    }
    if user_id:
        request_body["user_id"] = user_id
    response = self._client.post(
        "/api/v1/user/symphonies/modify",
        json=request_body,
    )
    return response

list_drafts()

Get all draft symphonies in the user's library.

Returns a list of draft (unpublished) symphonies created by the authenticated user.

Returns

List[DraftSymphony]: List of user's draft symphonies with statistics.

Example

drafts = client.user_symphonies.list_drafts() for draft in drafts:

...     print(f"Draft: {draft.name}")

Source code in composer/resources/user_symphonies.py

def list_drafts(self) -> list[DraftSymphony]:
    """Get all draft symphonies in the user's library.

    Returns a list of draft (unpublished) symphonies created by the
    authenticated user.

    Returns
    -------
        List[DraftSymphony]: List of user's draft symphonies with statistics.

    Example:
         drafts = client.user_symphonies.list_drafts()
         for draft in drafts:
        ...     print(f"Draft: {draft.name}")

    """
    response = self._client.get("/api/v1/user/symphonies/drafts")
    result = DraftSymphoniesResponse.model_validate(response)
    return result.symphonies

list_symphonies()

Get all symphonies in the user's library.

Returns a list of all symphonies created by or copied to the authenticated user's account.

Returns

List[UserSymphony]: List of user's symphonies with statistics.

Example

symphonies = client.user_symphonies.list_symphonies() for symphony in symphonies:

...     print(f"{symphony.name}: Sharpe={symphony.oos_sharpe_ratio}")

Source code in composer/resources/user_symphonies.py

def list_symphonies(self) -> list[UserSymphony]:
    """Get all symphonies in the user's library.

    Returns a list of all symphonies created by or copied to the
    authenticated user's account.

    Returns
    -------
        List[UserSymphony]: List of user's symphonies with statistics.

    Example:
         symphonies = client.user_symphonies.list_symphonies()
         for symphony in symphonies:
        ...     print(f"{symphony.name}: Sharpe={symphony.oos_sharpe_ratio}")

    """
    response = self._client.get("/api/v1/user/symphonies")
    result = UserSymphoniesResponse.model_validate(response)
    return result.symphonies

pubsub_modify_symphonies(subscription, message)

Programmatically modify all of a user's symphonies via pub/sub.

Parameters:

Name	Type	Description	Default
subscription	str	The pub/sub subscription name.	required
message	dict[str, Any]	The message containing modification instructions.	required

Returns

Dict[str, Any]: Response from the API.

Example

result = client.user_symphonies.pubsub_modify_symphonies(

...     subscription="my-subscription",
...     message={"publish_time": "2024-01-01", "data": {...}}
... )
 print(result)

Source code in composer/resources/user_symphonies.py

def pubsub_modify_symphonies(
    self,
    subscription: str,
    message: dict[str, Any],
) -> dict[str, Any]:
    """Programmatically modify all of a user's symphonies via pub/sub.

    Args:
        subscription: The pub/sub subscription name.
        message: The message containing modification instructions.

    Returns
    -------
        Dict[str, Any]: Response from the API.

    Example:
         result = client.user_symphonies.pubsub_modify_symphonies(
        ...     subscription="my-subscription",
        ...     message={"publish_time": "2024-01-01", "data": {...}}
        ... )
         print(result)

    """
    request_body = {
        "subscription": subscription,
        "message": message,
    }
    response = self._client.post(
        "/api/v1/pubsub/symphonies/modify",
        json=request_body,
    )
    return response

Search Resource

composer.resources.search.Search

Resource for searching publicly shared symphonies.

Source code in composer/resources/search.py

class Search:
    """Resource for searching publicly shared symphonies."""

    def __init__(self, http_client):
        self.http_client = http_client

    def search_symphonies(
        self,
        where: list[Any] | None = None,
        order_by: list[list[str]] | None = None,
        offset: int = 0,
    ) -> list[SearchSymphonyResult]:
        """
        Search our database of publicly shared symphonies.

        Args:
            where: Filter conditions for the search. Uses HoneySQL (Clojure) syntax.
                   Format: ["and", [condition1], [condition2], ...]
                   Example: ["and", [">", "oos_num_backtest_days", 180]]
            order_by: Sorting criteria as list of [field, direction] pairs.
                      For example: [["oos_sharpe_ratio", "desc"]]
            offset: Pagination offset for results

        Returns
        -------
            List of matching symphonies with performance statistics
        """
        if where is None:
            where = []
        request: dict[str, Any] = {"offset": offset}
        if where is not None:
            request["where"] = where
        if order_by is not None:
            request["order_by"] = order_by

        response = self.http_client.post(
            "/api/v1/public/search/symphonies",
            json=request,
        )

        return [SearchSymphonyResult.model_validate(r) for r in response]

    def search_symphonies_v2(
        self,
        filter: str | None = None,
        order_by: list[list[str]] | None = None,
        offset: int = 0,
    ) -> list[SearchSymphonyResult]:
        """
        Search publicly shared symphonies using CEL filters (V2).

        Args:
            filter: CEL filter expression for advanced filtering.
                    Example: "oos_sharpe_ratio > 1.5"
            order_by: Sorting criteria as list of [field, direction] pairs.
                      For example: [["oos_sharpe_ratio", "desc"]]
            offset: Pagination offset for results

        Returns
        -------
            List of matching symphonies with performance statistics
        """
        request: dict[str, Any] = {"offset": offset}
        if filter is not None:
            request["filter"] = filter
        if order_by is not None:
            request["order_by"] = order_by

        response = self.http_client.post(
            "/api/v1/public/search/symphonies-v2",
            json=request,
        )
        return [SearchSymphonyResult.model_validate(r) for r in response]

search_symphonies(where=None, order_by=None, offset=0)

Search our database of publicly shared symphonies.

Parameters:

Name	Type	Description	Default
where	list[Any] | None	Filter conditions for the search. Uses HoneySQL (Clojure) syntax. Format: ["and", [condition1], [condition2], ...] Example: ["and", [">", "oos_num_backtest_days", 180]]	None
order_by	list[list[str]] | None	Sorting criteria as list of [field, direction] pairs. For example: [["oos_sharpe_ratio", "desc"]]	None
offset	int	Pagination offset for results	0

Returns

List of matching symphonies with performance statistics

Source code in composer/resources/search.py

def search_symphonies(
    self,
    where: list[Any] | None = None,
    order_by: list[list[str]] | None = None,
    offset: int = 0,
) -> list[SearchSymphonyResult]:
    """
    Search our database of publicly shared symphonies.

    Args:
        where: Filter conditions for the search. Uses HoneySQL (Clojure) syntax.
               Format: ["and", [condition1], [condition2], ...]
               Example: ["and", [">", "oos_num_backtest_days", 180]]
        order_by: Sorting criteria as list of [field, direction] pairs.
                  For example: [["oos_sharpe_ratio", "desc"]]
        offset: Pagination offset for results

    Returns
    -------
        List of matching symphonies with performance statistics
    """
    if where is None:
        where = []
    request: dict[str, Any] = {"offset": offset}
    if where is not None:
        request["where"] = where
    if order_by is not None:
        request["order_by"] = order_by

    response = self.http_client.post(
        "/api/v1/public/search/symphonies",
        json=request,
    )

    return [SearchSymphonyResult.model_validate(r) for r in response]

search_symphonies_v2(filter=None, order_by=None, offset=0)

Search publicly shared symphonies using CEL filters (V2).

Parameters:

Name	Type	Description	Default
filter	str | None	CEL filter expression for advanced filtering. Example: "oos_sharpe_ratio > 1.5"	None
order_by	list[list[str]] | None	Sorting criteria as list of [field, direction] pairs. For example: [["oos_sharpe_ratio", "desc"]]	None
offset	int	Pagination offset for results	0

Returns

List of matching symphonies with performance statistics

Source code in composer/resources/search.py

def search_symphonies_v2(
    self,
    filter: str | None = None,
    order_by: list[list[str]] | None = None,
    offset: int = 0,
) -> list[SearchSymphonyResult]:
    """
    Search publicly shared symphonies using CEL filters (V2).

    Args:
        filter: CEL filter expression for advanced filtering.
                Example: "oos_sharpe_ratio > 1.5"
        order_by: Sorting criteria as list of [field, direction] pairs.
                  For example: [["oos_sharpe_ratio", "desc"]]
        offset: Pagination offset for results

    Returns
    -------
        List of matching symphonies with performance statistics
    """
    request: dict[str, Any] = {"offset": offset}
    if filter is not None:
        request["filter"] = filter
    if order_by is not None:
        request["order_by"] = order_by

    response = self.http_client.post(
        "/api/v1/public/search/symphonies-v2",
        json=request,
    )
    return [SearchSymphonyResult.model_validate(r) for r in response]

Watchlist Resource

composer.resources.watchlist.Watchlist

Watchlist endpoints.

These endpoints provide access to the user's watchlist of symphonies they are tracking but haven't necessarily invested in.

Source code in composer/resources/watchlist.py

class Watchlist:
    """
    Watchlist endpoints.

    These endpoints provide access to the user's watchlist of symphonies
    they are tracking but haven't necessarily invested in.
    """

    def __init__(self, http_client: HTTPClient):
        self._client = http_client

    def get_watchlist(self) -> list[WatchlistSymphony]:
        """
        Get all symphonies on the user's watchlist.

        Returns a list of symphonies the user has added to their watchlist
        for tracking purposes.

        Returns
        -------
            List[WatchlistSymphony]: List of watchlist symphonies with statistics.

        Example:
             watchlist = client.watchlist.get_watchlist()
             for item in watchlist:
            ...     print(f"Watching: {item.name}")
            ...     print(f"  Sharpe: {item.oos_sharpe_ratio}")
        """
        response = self._client.get("/api/v1/watchlist")
        result = WatchlistResponse.model_validate(response)
        return result.watchlist or [] or []

    def add_to_watchlist(self, symphony_id: str) -> WatchlistSymphonyItem:
        """
        Add a symphony to the user's watchlist.

        Args:
            symphony_id: The unique identifier of the symphony to add.

        Returns
        -------
            WatchlistSymphonyItem: The added symphony with its details.

        Example:
             result = client.watchlist.add_to_watchlist("fk6VGRDAAgiH120TfUPS")
             print(f"Added: {result.name}")
        """
        response = self._client.post(f"/api/v1/watchlist/{symphony_id}")
        return WatchlistSymphonyItem.model_validate(response)

    def remove_from_watchlist(self, symphony_id: str) -> None:
        """
        Remove a symphony from the user's watchlist.

        Args:
            symphony_id: The unique identifier of the symphony to remove.

        Example:
             client.watchlist.remove_from_watchlist("fk6VGRDAAgiH120TfUPS")
        """
        self._client.delete(f"/api/v1/watchlist/{symphony_id}")

add_to_watchlist(symphony_id)

Add a symphony to the user's watchlist.

Parameters:

Name	Type	Description	Default
symphony_id	str	The unique identifier of the symphony to add.	required

Returns

WatchlistSymphonyItem: The added symphony with its details.

Example

result = client.watchlist.add_to_watchlist("fk6VGRDAAgiH120TfUPS") print(f"Added: {result.name}")

Source code in composer/resources/watchlist.py

def add_to_watchlist(self, symphony_id: str) -> WatchlistSymphonyItem:
    """
    Add a symphony to the user's watchlist.

    Args:
        symphony_id: The unique identifier of the symphony to add.

    Returns
    -------
        WatchlistSymphonyItem: The added symphony with its details.

    Example:
         result = client.watchlist.add_to_watchlist("fk6VGRDAAgiH120TfUPS")
         print(f"Added: {result.name}")
    """
    response = self._client.post(f"/api/v1/watchlist/{symphony_id}")
    return WatchlistSymphonyItem.model_validate(response)

get_watchlist()

Get all symphonies on the user's watchlist.

Returns a list of symphonies the user has added to their watchlist for tracking purposes.

Returns

List[WatchlistSymphony]: List of watchlist symphonies with statistics.

Example

watchlist = client.watchlist.get_watchlist() for item in watchlist:

...     print(f"Watching: {item.name}")
...     print(f"  Sharpe: {item.oos_sharpe_ratio}")

Source code in composer/resources/watchlist.py

def get_watchlist(self) -> list[WatchlistSymphony]:
    """
    Get all symphonies on the user's watchlist.

    Returns a list of symphonies the user has added to their watchlist
    for tracking purposes.

    Returns
    -------
        List[WatchlistSymphony]: List of watchlist symphonies with statistics.

    Example:
         watchlist = client.watchlist.get_watchlist()
         for item in watchlist:
        ...     print(f"Watching: {item.name}")
        ...     print(f"  Sharpe: {item.oos_sharpe_ratio}")
    """
    response = self._client.get("/api/v1/watchlist")
    result = WatchlistResponse.model_validate(response)
    return result.watchlist or [] or []

remove_from_watchlist(symphony_id)

Remove a symphony from the user's watchlist.

Parameters:

Name	Type	Description	Default
symphony_id	str	The unique identifier of the symphony to remove.	required

Example

client.watchlist.remove_from_watchlist("fk6VGRDAAgiH120TfUPS")

Source code in composer/resources/watchlist.py

def remove_from_watchlist(self, symphony_id: str) -> None:
    """
    Remove a symphony from the user's watchlist.

    Args:
        symphony_id: The unique identifier of the symphony to remove.

    Example:
         client.watchlist.remove_from_watchlist("fk6VGRDAAgiH120TfUPS")
    """
    self._client.delete(f"/api/v1/watchlist/{symphony_id}")