Portfolio Management Guide¶

from tradedesk.portfolio import PortfolioRunner, EqualSplitRiskPolicy, Instrument

# Create strategies for different instruments
strategies = {
    Instrument("EURUSD"): eurusd_strategy,
    Instrument("GBPUSD"): gbpusd_strategy,
    Instrument("USDJPY"): usdjpy_strategy,
}

# Create portfolio runner with risk policy
runner = PortfolioRunner(
    strategies=strategies,
    policy=EqualSplitRiskPolicy(portfolio_risk_budget=100.0),
    default_risk_per_trade=10.0,
)

# Process candle events
await runner.on_candle_close(candle_event)

from tradedesk.portfolio import PortfolioStrategy, Instrument
from tradedesk.marketdata.events import CandleClosedEvent

class MyStrategy:
    def __init__(self, instrument: str):
        self.instrument = Instrument(instrument)
        self._risk_per_trade = 10.0
        self._regime_active = False

    def set_risk_per_trade(self, value: float) -> None:
        """Called by PortfolioRunner between phase 1 and 2."""
        self._risk_per_trade = value

    def is_regime_active(self) -> bool:
        """Return True if strategy's regime is active."""
        return self._regime_active

    async def update_state(self, event: CandleClosedEvent) -> None:
        """Phase 1: Update indicators and regime state.

        Do NOT make trading decisions here. This happens before
        risk allocation, so risk_per_trade may not be current.
        """
        # Update indicators, regime filters, position tracking
        candle = event.candle
        # ... update logic ...
        self._regime_active = self._check_regime(candle)

    async def evaluate_signals(self) -> None:
        """Phase 2: Evaluate signals and execute trades.

        Make trading decisions here. This happens after risk
        allocation, so risk_per_trade is current and correct.
        """
        # Check entry/exit conditions and place orders
        # ... trading logic using self._risk_per_trade ...
        pass

from tradedesk.portfolio import EqualSplitRiskPolicy, Instrument

policy = EqualSplitRiskPolicy(portfolio_risk_budget=100.0)

# If 2 instruments are active, each gets 50.0
active = [Instrument("EURUSD"), Instrument("GBPUSD")]
allocation = policy.allocate(active)
# Returns: {Instrument("EURUSD"): 50.0, Instrument("GBPUSD"): 50.0}

# If 0 instruments are active, returns empty dict
allocation = policy.allocate([])
# Returns: {}

from dataclasses import dataclass
from typing import Mapping
from tradedesk.portfolio.types import Instrument

@dataclass(frozen=True)
class VolatilityWeightedPolicy:
    """Allocate more risk to less volatile instruments."""
    portfolio_risk_budget: float
    volatilities: dict[Instrument, float]

    def allocate(self, active_instruments: list[Instrument]) -> Mapping[Instrument, float]:
        if not active_instruments:
            return {}

        # Calculate inverse volatility weights
        inv_vols = {inst: 1.0 / self.volatilities[inst] for inst in active_instruments}
        total_weight = sum(inv_vols.values())

        # Allocate proportionally
        return {
            inst: (inv_vols[inst] / total_weight) * self.portfolio_risk_budget
            for inst in active_instruments
        }

await strategy.update_state(event)

self._apply_risk_budgets()

await strategy.evaluate_signals()

from tradedesk.portfolio import (
    PortfolioRunner,
    EqualSplitRiskPolicy,
    Instrument,
)
from tradedesk import Candle
from tradedesk.marketdata.events import CandleClosedEvent

class SimpleStrategy:
    def __init__(self, instrument: str, threshold: float):
        self.instrument = Instrument(instrument)
        self.threshold = threshold
        self._risk_per_trade = 10.0
        self._regime_active = False
        self._price_history = []
        self._current_candle = None

    def set_risk_per_trade(self, value: float) -> None:
        self._risk_per_trade = value

    def is_regime_active(self) -> bool:
        return self._regime_active

    async def update_state(self, event: CandleClosedEvent) -> None:
        """Phase 1: Update indicators and regime state."""
        self._current_candle = event.candle
        self._price_history.append(event.candle.close)

        # Simple volatility regime: activate if price moves > threshold
        if len(self._price_history) >= 2:
            move = abs(event.candle.close - self._price_history[-2])
            self._regime_active = move > self.threshold

    async def evaluate_signals(self) -> None:
        """Phase 2: Make trading decisions with correct risk allocation."""
        if not self._regime_active or not self._current_candle:
            return

        # Entry logic using self._risk_per_trade (which is now current)
        # ... trading decisions here ...

# Create strategies
strategies = {
    Instrument("EURUSD"): SimpleStrategy("EURUSD", threshold=0.001),
    Instrument("GBPUSD"): SimpleStrategy("GBPUSD", threshold=0.002),
}

# Create runner
runner = PortfolioRunner(
    strategies=strategies,
    policy=EqualSplitRiskPolicy(portfolio_risk_budget=100.0),
    default_risk_per_trade=50.0,
)

# Process events
candle = Candle(timestamp="1234567890000", open=1.1000, high=1.1050,
                low=1.0950, close=1.1020)
await runner.on_candle_close(
    CandleClosedEvent(
        instrument=Instrument("EURUSD"),
        timeframe="15MINUTE",
        candle=candle,
    )
)