Risk calculation exercises¶

Explore account metrics, an illustrative FX sizing calculation and dependent-order helpers. Example percentages and scores are application settings, not recommended risk limits or calibrated measures of account safety. Calling the order helpers changes account state; inspect each call before running it.

Setup¶

Use a dedicated OANDA practice account and set both FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT in the kernel's environment. Review the configuration cell before making requests. Install the packages imported by the setup cell, select that Python environment as the Jupyter kernel, and run cells in order.

The repository's uv run poe docs-validate-notebooks command executes a temporary copy with synthetic HTTP responses. Running this notebook normally uses its configured API credentials; the offline validation result is not a live account test.

Setup and Imports¶

In [ ]:

import os
from datetime import datetime, timezone
from decimal import Decimal

from fivetwenty import AsyncClient, Environment
from fivetwenty.exceptions import FiveTwentyError
from fivetwenty.models import LimitOrderRequest, OrderPositionFill, TimeInForce, TrailingStopLossDetails

# Jupyter async support
try:
    import nest_asyncio

    nest_asyncio.apply()
except ImportError:
    print("Install nest_asyncio for Jupyter: uv add nest_asyncio")

# Configuration
TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-token-here")
ENVIRONMENT = Environment.PRACTICE
ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here")

print("✅ Setup complete" if TOKEN != "your-token-here" and ACCOUNT_ID != "your-account-id-here" else "⚠️ Set FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT environment variables")

import os from datetime import datetime, timezone from decimal import Decimal from fivetwenty import AsyncClient, Environment from fivetwenty.exceptions import FiveTwentyError from fivetwenty.models import LimitOrderRequest, OrderPositionFill, TimeInForce, TrailingStopLossDetails # Jupyter async support try: import nest_asyncio nest_asyncio.apply() except ImportError: print("Install nest_asyncio for Jupyter: uv add nest_asyncio") # Configuration TOKEN = os.getenv("FIVETWENTY_OANDA_TOKEN", "your-token-here") ENVIRONMENT = Environment.PRACTICE ACCOUNT_ID = os.getenv("FIVETWENTY_OANDA_ACCOUNT", "your-account-id-here") print("✅ Setup complete" if TOKEN != "your-token-here" and ACCOUNT_ID != "your-account-id-here" else "⚠️ Set FIVETWENTY_OANDA_TOKEN and FIVETWENTY_OANDA_ACCOUNT environment variables")

Risk Management Framework¶

The following class calculates example metrics and sizing estimates. Its limits and scores are illustrative application policy.

In [ ]:

class RiskManager:
    """Comprehensive risk management framework."""

    def __init__(self, client: AsyncClient, account_id: str):
        self.client = client
        self.account_id = account_id

        # Risk parameters (customize these based on your risk tolerance)
        self.max_risk_per_trade = Decimal("0.02")  # 2% of account per trade
        self.max_daily_loss = Decimal("0.05")  # 5% maximum daily loss
        self.max_portfolio_risk = Decimal("0.10")  # 10% maximum portfolio risk
        self.max_drawdown = Decimal("0.15")  # 15% maximum drawdown

        # Position limits
        self.max_positions = 5  # Maximum number of open positions
        self.max_correlation = Decimal("0.7")  # Maximum correlation between positions

    async def get_account_info(self) -> dict:
        """Get current account information."""
        try:
            account_response = await self.client.accounts.get_account(self.account_id)
            account = account_response["account"]
            return {
                "balance": Decimal(str(account.balance)),
                "nav": Decimal(str(account.nav)),
                "unrealized_pl": Decimal(str(account.unrealized_pl)),
                "margin_used": Decimal(str(account.margin_used)),
                "margin_available": Decimal(str(account.margin_available)),
                "open_trade_count": int(account.open_trade_count),
                "open_position_count": int(account.open_position_count),
                "currency": account.currency,
            }
        except FiveTwentyError as e:
            print(f"Error getting account info: {e.message}")
            return {}

    async def calculate_position_size(self, instrument: str, entry_price: Decimal, stop_loss: Decimal, risk_amount: Decimal | None = None) -> int:
        """Estimate whole units at the specified stop price, excluding slippage and fees."""

        account_info = await self.get_account_info()
        if not account_info:
            return 0

        account_balance = account_info["balance"]

        # Use specified risk amount or default to max risk per trade
        if risk_amount is None:
            risk_amount = account_balance * self.max_risk_per_trade

        if not risk_amount.is_finite() or risk_amount <= 0:
            raise ValueError("Risk amount must be finite and positive")
        if not entry_price.is_finite() or not stop_loss.is_finite() or entry_price <= 0 or stop_loss <= 0:
            raise ValueError("Entry and stop prices must be finite and positive")

        # Price distance is already the loss in quote currency per unit.
        # Convert that loss into the account's home currency before sizing.
        price_distance = abs(entry_price - stop_loss)
        if price_distance == 0:
            print("⚠️ Warning: Entry price equals stop loss price")
            return 0

        loss_conversion = await self.get_loss_conversion_factor(instrument, account_info["currency"])
        loss_per_unit = price_distance * loss_conversion
        position_size = int(risk_amount / loss_per_unit)  # Round down to whole units.

        print("📊 Position Size Calculation:")
        print(f"  Risk Amount: {risk_amount:.2f} {account_info['currency']}")
        print(f"  Price Distance: {price_distance}")
        print(f"  Loss per Unit: {loss_per_unit} {account_info['currency']}")
        print(f"  Calculated Size: {position_size} units")

        return position_size

    async def get_loss_conversion_factor(self, instrument: str, account_currency: str) -> Decimal:
        """Convert one unit of the instrument's quote currency into home-currency loss."""
        if "_" not in instrument:
            raise ValueError("Expected an instrument with a quote currency, such as EUR_USD")
        quote_currency = instrument.rsplit("_", 1)[1]
        if quote_currency == account_currency:
            return Decimal("1")

        pricing = await self.client.pricing.get_pricing(
            account_id=self.account_id,
            instruments=[instrument],
            include_home_conversions=True,
        )
        for conversion in pricing.get("homeConversions", []):
            if conversion.currency == quote_currency:
                factor = conversion.account_loss
                if factor.is_finite() and factor > 0:
                    return factor
                break
        raise ValueError(f"No valid loss conversion from {quote_currency} to {account_currency}; position sizing unavailable")

    async def check_daily_loss_limit(self) -> bool:
        """Check if daily loss limit has been exceeded."""
        today = datetime.now(timezone.utc).date()

        try:
            response = await self.client.transactions.get_recent_transactions(account_id=self.account_id, count=500)

            # Calculate today's P/L
            daily_pl = Decimal("0")
            for transaction in response["transactions"]:
                if transaction.time.date() != today:
                    continue
                if getattr(transaction, "pl", None):
                    daily_pl += Decimal(str(transaction.pl))

            account_info = await self.get_account_info()
            account_balance = account_info["balance"]

            max_daily_loss_amount = account_balance * self.max_daily_loss

            print("📊 Daily Loss Check:")
            print(f"  Today's P/L: {daily_pl:.2f}")
            print(f"  Max Daily Loss: {max_daily_loss_amount:.2f}")

            if abs(daily_pl) > max_daily_loss_amount:
                print("🚨 Daily loss limit exceeded!")
                return False

            return True

        except FiveTwentyError as e:
            print(f"Error checking daily loss: {e.message}")
            return True  # Allow trading if we can't check

    async def check_portfolio_risk(self) -> dict:
        """Assess overall portfolio risk."""
        try:
            positions_response = await self.client.positions.get_open_positions(self.account_id)
            positions = positions_response["positions"]
            account_info = await self.get_account_info()

            total_risk = Decimal("0")
            position_details = []

            for position in positions:
                # Calculate risk for each position
                if position.long.units != 0:
                    unrealized_pl = Decimal(str(position.long.unrealized_pl))
                    units = int(position.long.units)
                elif position.short.units != 0:
                    unrealized_pl = Decimal(str(position.short.unrealized_pl))
                    units = int(position.short.units)
                else:
                    continue

                position_risk = abs(unrealized_pl) / account_info["balance"]
                total_risk += position_risk

                position_details.append({"instrument": position.instrument, "units": units, "unrealized_pl": unrealized_pl, "risk_percentage": position_risk * Decimal("100")})

            risk_assessment = {
                "total_risk_percentage": total_risk * Decimal("100"),
                "max_portfolio_risk_percentage": self.max_portfolio_risk * Decimal("100"),
                "within_limits": total_risk <= self.max_portfolio_risk,
                "position_count": len(positions),
                "max_positions": self.max_positions,
                "position_details": position_details,
            }

            return risk_assessment

        except FiveTwentyError as e:
            print(f"Error assessing portfolio risk: {e.message}")
            return {}

    async def validate_trade(self, instrument: str, units: int, entry_price: Decimal, stop_loss: Decimal, take_profit: Decimal | None = None) -> dict:
        """Comprehensive trade validation before execution."""

        validation_result = {"approved": True, "warnings": [], "errors": [], "adjustments": {}}

        # Check daily loss limit
        if not await self.check_daily_loss_limit():
            validation_result["approved"] = False
            validation_result["errors"].append("Daily loss limit exceeded")

        # Check portfolio risk
        portfolio_risk = await self.check_portfolio_risk()
        if portfolio_risk and not portfolio_risk["within_limits"]:
            validation_result["warnings"].append(f"Portfolio risk ({portfolio_risk['total_risk_percentage']:.1f}%) exceeds limit")

        # Check position count
        if portfolio_risk and portfolio_risk["position_count"] >= self.max_positions:
            validation_result["approved"] = False
            validation_result["errors"].append(f"Maximum positions ({self.max_positions}) already reached")

        # Validate position size
        optimal_size = await self.calculate_position_size(instrument, entry_price, stop_loss)
        if abs(units) > optimal_size * Decimal("1.5"):  # Allow 50% variance
            validation_result["warnings"].append(f"Position size ({abs(units)}) exceeds optimal size ({optimal_size})")
            validation_result["adjustments"]["suggested_units"] = optimal_size if units > 0 else -optimal_size

        # Risk-reward ratio check
        if take_profit:
            risk = abs(entry_price - stop_loss)
            reward = abs(take_profit - entry_price)
            risk_reward_ratio = reward / risk if risk > 0 else Decimal("0")

            if risk_reward_ratio < Decimal("1.5"):  # Minimum 1.5:1 risk-reward
                validation_result["warnings"].append(f"Risk-reward ratio ({risk_reward_ratio:.2f}) below recommended 1.5:1")

        return validation_result


print("✅ Risk management framework defined")

class RiskManager: """Comprehensive risk management framework.""" def __init__(self, client: AsyncClient, account_id: str): self.client = client self.account_id = account_id # Risk parameters (customize these based on your risk tolerance) self.max_risk_per_trade = Decimal("0.02") # 2% of account per trade self.max_daily_loss = Decimal("0.05") # 5% maximum daily loss self.max_portfolio_risk = Decimal("0.10") # 10% maximum portfolio risk self.max_drawdown = Decimal("0.15") # 15% maximum drawdown # Position limits self.max_positions = 5 # Maximum number of open positions self.max_correlation = Decimal("0.7") # Maximum correlation between positions async def get_account_info(self) -> dict: """Get current account information.""" try: account_response = await self.client.accounts.get_account(self.account_id) account = account_response["account"] return { "balance": Decimal(str(account.balance)), "nav": Decimal(str(account.nav)), "unrealized_pl": Decimal(str(account.unrealized_pl)), "margin_used": Decimal(str(account.margin_used)), "margin_available": Decimal(str(account.margin_available)), "open_trade_count": int(account.open_trade_count), "open_position_count": int(account.open_position_count), "currency": account.currency, } except FiveTwentyError as e: print(f"Error getting account info: {e.message}") return {} async def calculate_position_size(self, instrument: str, entry_price: Decimal, stop_loss: Decimal, risk_amount: Decimal | None = None) -> int: """Estimate whole units at the specified stop price, excluding slippage and fees.""" account_info = await self.get_account_info() if not account_info: return 0 account_balance = account_info["balance"] # Use specified risk amount or default to max risk per trade if risk_amount is None: risk_amount = account_balance * self.max_risk_per_trade if not risk_amount.is_finite() or risk_amount <= 0: raise ValueError("Risk amount must be finite and positive") if not entry_price.is_finite() or not stop_loss.is_finite() or entry_price <= 0 or stop_loss <= 0: raise ValueError("Entry and stop prices must be finite and positive") # Price distance is already the loss in quote currency per unit. # Convert that loss into the account's home currency before sizing. price_distance = abs(entry_price - stop_loss) if price_distance == 0: print("⚠️ Warning: Entry price equals stop loss price") return 0 loss_conversion = await self.get_loss_conversion_factor(instrument, account_info["currency"]) loss_per_unit = price_distance * loss_conversion position_size = int(risk_amount / loss_per_unit) # Round down to whole units. print("📊 Position Size Calculation:") print(f" Risk Amount: {risk_amount:.2f} {account_info['currency']}") print(f" Price Distance: {price_distance}") print(f" Loss per Unit: {loss_per_unit} {account_info['currency']}") print(f" Calculated Size: {position_size} units") return position_size async def get_loss_conversion_factor(self, instrument: str, account_currency: str) -> Decimal: """Convert one unit of the instrument's quote currency into home-currency loss.""" if "_" not in instrument: raise ValueError("Expected an instrument with a quote currency, such as EUR_USD") quote_currency = instrument.rsplit("_", 1)[1] if quote_currency == account_currency: return Decimal("1") pricing = await self.client.pricing.get_pricing( account_id=self.account_id, instruments=[instrument], include_home_conversions=True, ) for conversion in pricing.get("homeConversions", []): if conversion.currency == quote_currency: factor = conversion.account_loss if factor.is_finite() and factor > 0: return factor break raise ValueError(f"No valid loss conversion from {quote_currency} to {account_currency}; position sizing unavailable") async def check_daily_loss_limit(self) -> bool: """Check if daily loss limit has been exceeded.""" today = datetime.now(timezone.utc).date() try: response = await self.client.transactions.get_recent_transactions(account_id=self.account_id, count=500) # Calculate today's P/L daily_pl = Decimal("0") for transaction in response["transactions"]: if transaction.time.date() != today: continue if getattr(transaction, "pl", None): daily_pl += Decimal(str(transaction.pl)) account_info = await self.get_account_info() account_balance = account_info["balance"] max_daily_loss_amount = account_balance * self.max_daily_loss print("📊 Daily Loss Check:") print(f" Today's P/L: {daily_pl:.2f}") print(f" Max Daily Loss: {max_daily_loss_amount:.2f}") if abs(daily_pl) > max_daily_loss_amount: print("🚨 Daily loss limit exceeded!") return False return True except FiveTwentyError as e: print(f"Error checking daily loss: {e.message}") return True # Allow trading if we can't check async def check_portfolio_risk(self) -> dict: """Assess overall portfolio risk.""" try: positions_response = await self.client.positions.get_open_positions(self.account_id) positions = positions_response["positions"] account_info = await self.get_account_info() total_risk = Decimal("0") position_details = [] for position in positions: # Calculate risk for each position if position.long.units != 0: unrealized_pl = Decimal(str(position.long.unrealized_pl)) units = int(position.long.units) elif position.short.units != 0: unrealized_pl = Decimal(str(position.short.unrealized_pl)) units = int(position.short.units) else: continue position_risk = abs(unrealized_pl) / account_info["balance"] total_risk += position_risk position_details.append({"instrument": position.instrument, "units": units, "unrealized_pl": unrealized_pl, "risk_percentage": position_risk * Decimal("100")}) risk_assessment = { "total_risk_percentage": total_risk * Decimal("100"), "max_portfolio_risk_percentage": self.max_portfolio_risk * Decimal("100"), "within_limits": total_risk <= self.max_portfolio_risk, "position_count": len(positions), "max_positions": self.max_positions, "position_details": position_details, } return risk_assessment except FiveTwentyError as e: print(f"Error assessing portfolio risk: {e.message}") return {} async def validate_trade(self, instrument: str, units: int, entry_price: Decimal, stop_loss: Decimal, take_profit: Decimal | None = None) -> dict: """Comprehensive trade validation before execution.""" validation_result = {"approved": True, "warnings": [], "errors": [], "adjustments": {}} # Check daily loss limit if not await self.check_daily_loss_limit(): validation_result["approved"] = False validation_result["errors"].append("Daily loss limit exceeded") # Check portfolio risk portfolio_risk = await self.check_portfolio_risk() if portfolio_risk and not portfolio_risk["within_limits"]: validation_result["warnings"].append(f"Portfolio risk ({portfolio_risk['total_risk_percentage']:.1f}%) exceeds limit") # Check position count if portfolio_risk and portfolio_risk["position_count"] >= self.max_positions: validation_result["approved"] = False validation_result["errors"].append(f"Maximum positions ({self.max_positions}) already reached") # Validate position size optimal_size = await self.calculate_position_size(instrument, entry_price, stop_loss) if abs(units) > optimal_size * Decimal("1.5"): # Allow 50% variance validation_result["warnings"].append(f"Position size ({abs(units)}) exceeds optimal size ({optimal_size})") validation_result["adjustments"]["suggested_units"] = optimal_size if units > 0 else -optimal_size # Risk-reward ratio check if take_profit: risk = abs(entry_price - stop_loss) reward = abs(take_profit - entry_price) risk_reward_ratio = reward / risk if risk > 0 else Decimal("0") if risk_reward_ratio < Decimal("1.5"): # Minimum 1.5:1 risk-reward validation_result["warnings"].append(f"Risk-reward ratio ({risk_reward_ratio:.2f}) below recommended 1.5:1") return validation_result print("✅ Risk management framework defined")

Advanced Order Types for Risk Management¶

Let's create functions for advanced order types that help manage risk:

In [ ]:

class AdvancedOrders:
    """Advanced order management for risk control."""

    def __init__(self, client: AsyncClient, account_id: str):
        self.client = client
        self.account_id = account_id

    async def place_bracket_order(self, instrument: str, units: int, stop_loss_pips: float, take_profit_pips: float) -> dict:
        """Place a market order with automatic stop loss and take profit."""

        try:
            # Get current price
            pricing_response = await self.client.pricing.get_pricing(account_id=self.account_id, instruments=[instrument])
            prices = pricing_response["prices"]

            if not prices or not prices[0].asks or not prices[0].bids:
                return {"success": False, "error": "Could not get current price"}

            # Determine entry price based on direction
            if units > 0:  # Buy order
                entry_price = float(prices[0].asks[0].price)
                stop_loss_price = entry_price - (stop_loss_pips * 0.0001)
                take_profit_price = entry_price + (take_profit_pips * 0.0001)
            else:  # Sell order
                entry_price = float(prices[0].bids[0].price)
                stop_loss_price = entry_price + (stop_loss_pips * 0.0001)
                take_profit_price = entry_price - (take_profit_pips * 0.0001)

            # Adjust for JPY pairs
            if instrument.endswith("JPY"):
                stop_loss_price = entry_price + (stop_loss_pips * 0.01 * (-1 if units > 0 else 1))
                take_profit_price = entry_price + (take_profit_pips * 0.01 * (1 if units > 0 else -1))

            # Place bracket order
            response = await self.client.orders.post_market_order(account_id=self.account_id, instrument=instrument, units=units, stop_loss=Decimal(f"{stop_loss_price:.5f}"), take_profit=Decimal(f"{take_profit_price:.5f}"))

            if response.order_fill_transaction:
                fill = response.order_fill_transaction
                return {"success": True, "trade_id": fill.trade_opened.trade_id if fill.trade_opened else None, "fill_price": float(fill.price), "stop_loss": stop_loss_price, "take_profit": take_profit_price, "units": int(fill.units)}
            return {"success": False, "error": "Order not filled"}

        except FiveTwentyError as e:
            return {"success": False, "error": f"OANDA error: {e.message}"}

    async def trailing_stop_loss(self, trade_id: str, trail_distance_pips: float) -> bool:
        """Set up a trailing stop loss for an existing trade."""

        try:
            # Get trade details
            trade = (await self.client.trades.get_trade(self.account_id, trade_id))["trade"]

            if not trade:
                print(f"❌ Trade {trade_id} not found")
                return False

            # Calculate trailing distance
            trail_distance = Decimal(str(trail_distance_pips * 0.01 if trade.instrument.endswith("JPY") else trail_distance_pips * 0.0001))

            # Update stop loss with trailing distance
            await self.client.trades.put_trade_orders(account_id=self.account_id, trade_specifier=trade_id, trailing_stop_loss=TrailingStopLossDetails(distance=trail_distance))

            print(f"✅ Trailing stop loss set: {trail_distance_pips} pips")
            return True

        except FiveTwentyError as e:
            print(f"❌ Error setting trailing stop: {e.message}")
            return False

    async def scale_out_position(self, trade_id: str, scale_percentages: list[float], price_targets: list[float]) -> list[dict]:
        """Create reduce-only profit limits for instrument exposure, sized from this trade.

        These resting orders are instrument-wide, not bound to the trade ID.
        Cancel outstanding limits if the position is closed another way.
        """

        if len(scale_percentages) != len(price_targets):
            print("❌ Scale percentages and price targets must have same length")
            return []

        try:
            # Get trade details
            trade = (await self.client.trades.get_trade(self.account_id, trade_id))["trade"]

            if not trade:
                print(f"❌ Trade {trade_id} not found")
                return []

            original_units = int(trade.current_units)
            scale_orders = []

            for i, (percentage, target_price) in enumerate(zip(scale_percentages, price_targets, strict=False)):
                # Calculate units to close
                units_to_close = int(abs(original_units) * percentage / 100)
                if original_units < 0:
                    units_to_close = -units_to_close

                if units_to_close == 0:
                    continue

                # Keep the price trigger while preventing an order from opening a hedge.
                request = LimitOrderRequest(
                    instrument=trade.instrument,
                    units=Decimal(-units_to_close),
                    price=Decimal(f"{target_price:.5f}"),
                    time_in_force=TimeInForce.GTC,
                    position_fill=OrderPositionFill.REDUCE_ONLY,
                )
                order_response = await self.client.orders.post_order(
                    account_id=self.account_id,
                    order_request=request,
                )

                if order_response.order_create_transaction:
                    order = order_response.order_create_transaction
                    scale_orders.append({"order_id": order.id, "percentage": percentage, "target_price": target_price, "units": units_to_close})
                    print(f"✅ Scale order {i + 1}: {percentage}% at {target_price}")

            return scale_orders

        except FiveTwentyError as e:
            print(f"❌ Error creating scale orders: {e.message}")
            return []


print("✅ Advanced orders class defined")

class AdvancedOrders: """Advanced order management for risk control.""" def __init__(self, client: AsyncClient, account_id: str): self.client = client self.account_id = account_id async def place_bracket_order(self, instrument: str, units: int, stop_loss_pips: float, take_profit_pips: float) -> dict: """Place a market order with automatic stop loss and take profit.""" try: # Get current price pricing_response = await self.client.pricing.get_pricing(account_id=self.account_id, instruments=[instrument]) prices = pricing_response["prices"] if not prices or not prices[0].asks or not prices[0].bids: return {"success": False, "error": "Could not get current price"} # Determine entry price based on direction if units > 0: # Buy order entry_price = float(prices[0].asks[0].price) stop_loss_price = entry_price - (stop_loss_pips * 0.0001) take_profit_price = entry_price + (take_profit_pips * 0.0001) else: # Sell order entry_price = float(prices[0].bids[0].price) stop_loss_price = entry_price + (stop_loss_pips * 0.0001) take_profit_price = entry_price - (take_profit_pips * 0.0001) # Adjust for JPY pairs if instrument.endswith("JPY"): stop_loss_price = entry_price + (stop_loss_pips * 0.01 * (-1 if units > 0 else 1)) take_profit_price = entry_price + (take_profit_pips * 0.01 * (1 if units > 0 else -1)) # Place bracket order response = await self.client.orders.post_market_order(account_id=self.account_id, instrument=instrument, units=units, stop_loss=Decimal(f"{stop_loss_price:.5f}"), take_profit=Decimal(f"{take_profit_price:.5f}")) if response.order_fill_transaction: fill = response.order_fill_transaction return {"success": True, "trade_id": fill.trade_opened.trade_id if fill.trade_opened else None, "fill_price": float(fill.price), "stop_loss": stop_loss_price, "take_profit": take_profit_price, "units": int(fill.units)} return {"success": False, "error": "Order not filled"} except FiveTwentyError as e: return {"success": False, "error": f"OANDA error: {e.message}"} async def trailing_stop_loss(self, trade_id: str, trail_distance_pips: float) -> bool: """Set up a trailing stop loss for an existing trade.""" try: # Get trade details trade = (await self.client.trades.get_trade(self.account_id, trade_id))["trade"] if not trade: print(f"❌ Trade {trade_id} not found") return False # Calculate trailing distance trail_distance = Decimal(str(trail_distance_pips * 0.01 if trade.instrument.endswith("JPY") else trail_distance_pips * 0.0001)) # Update stop loss with trailing distance await self.client.trades.put_trade_orders(account_id=self.account_id, trade_specifier=trade_id, trailing_stop_loss=TrailingStopLossDetails(distance=trail_distance)) print(f"✅ Trailing stop loss set: {trail_distance_pips} pips") return True except FiveTwentyError as e: print(f"❌ Error setting trailing stop: {e.message}") return False async def scale_out_position(self, trade_id: str, scale_percentages: list[float], price_targets: list[float]) -> list[dict]: """Create reduce-only profit limits for instrument exposure, sized from this trade. These resting orders are instrument-wide, not bound to the trade ID. Cancel outstanding limits if the position is closed another way. """ if len(scale_percentages) != len(price_targets): print("❌ Scale percentages and price targets must have same length") return [] try: # Get trade details trade = (await self.client.trades.get_trade(self.account_id, trade_id))["trade"] if not trade: print(f"❌ Trade {trade_id} not found") return [] original_units = int(trade.current_units) scale_orders = [] for i, (percentage, target_price) in enumerate(zip(scale_percentages, price_targets, strict=False)): # Calculate units to close units_to_close = int(abs(original_units) * percentage / 100) if original_units < 0: units_to_close = -units_to_close if units_to_close == 0: continue # Keep the price trigger while preventing an order from opening a hedge. request = LimitOrderRequest( instrument=trade.instrument, units=Decimal(-units_to_close), price=Decimal(f"{target_price:.5f}"), time_in_force=TimeInForce.GTC, position_fill=OrderPositionFill.REDUCE_ONLY, ) order_response = await self.client.orders.post_order( account_id=self.account_id, order_request=request, ) if order_response.order_create_transaction: order = order_response.order_create_transaction scale_orders.append({"order_id": order.id, "percentage": percentage, "target_price": target_price, "units": units_to_close}) print(f"✅ Scale order {i + 1}: {percentage}% at {target_price}") return scale_orders except FiveTwentyError as e: print(f"❌ Error creating scale orders: {e.message}") return [] print("✅ Advanced orders class defined")

Initialize Connection and Test Risk Management¶

In [ ]:

async def initialize_connection():
    """Initialize connection and get account ID."""
    global ACCOUNT_ID

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        try:
            accounts = await client.accounts.get_accounts()
            if accounts:
                ACCOUNT_ID = accounts[0].id
                print(f"✅ Connected to account: {ACCOUNT_ID}")
                return ACCOUNT_ID
            print("❌ No accounts found")
            return None
        except FiveTwentyError as e:
            print(f"❌ Connection error: {e.message}")
            return None


# Initialize connection
account_id = await initialize_connection()

async def initialize_connection(): """Initialize connection and get account ID.""" global ACCOUNT_ID async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: try: accounts = await client.accounts.get_accounts() if accounts: ACCOUNT_ID = accounts[0].id print(f"✅ Connected to account: {ACCOUNT_ID}") return ACCOUNT_ID print("❌ No accounts found") return None except FiveTwentyError as e: print(f"❌ Connection error: {e.message}") return None # Initialize connection account_id = await initialize_connection()

Portfolio Risk Assessment¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        # Initialize risk manager
        risk_manager = RiskManager(client, account_id)

        print("📊 Current Account Status:")
        account_info = await risk_manager.get_account_info()

        if account_info:
            print(f"  Balance: {account_info['balance']:.2f} {account_info['currency']}")
            print(f"  NAV: {account_info['nav']:.2f} {account_info['currency']}")
            print(f"  Unrealized P/L: {account_info['unrealized_pl']:.2f} {account_info['currency']}")
            print(f"  Margin Used: {account_info['margin_used']:.2f} {account_info['currency']}")
            print(f"  Open Positions: {account_info['open_position_count']}")
            print(f"  Open Trades: {account_info['open_trade_count']}")

        print("\n🔍 Portfolio Risk Assessment:")
        portfolio_risk = await risk_manager.check_portfolio_risk()

        if portfolio_risk:
            print(f"  Total Portfolio Risk: {portfolio_risk['total_risk_percentage']:.2f}%")
            print(f"  Risk Limit: {portfolio_risk['max_portfolio_risk_percentage']:.2f}%")
            print(f"  Within Limits: {'✅' if portfolio_risk['within_limits'] else '❌'}")

            if portfolio_risk["position_details"]:
                print("\n  Position Details:")
                for pos in portfolio_risk["position_details"]:
                    print(f"    {pos['instrument']}: {pos['units']} units, P/L: {pos['unrealized_pl']:.2f}, Risk: {pos['risk_percentage']:.2f}%")
else:
    print("❌ No account connection - cannot assess risk")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: # Initialize risk manager risk_manager = RiskManager(client, account_id) print("📊 Current Account Status:") account_info = await risk_manager.get_account_info() if account_info: print(f" Balance: {account_info['balance']:.2f} {account_info['currency']}") print(f" NAV: {account_info['nav']:.2f} {account_info['currency']}") print(f" Unrealized P/L: {account_info['unrealized_pl']:.2f} {account_info['currency']}") print(f" Margin Used: {account_info['margin_used']:.2f} {account_info['currency']}") print(f" Open Positions: {account_info['open_position_count']}") print(f" Open Trades: {account_info['open_trade_count']}") print("\n🔍 Portfolio Risk Assessment:") portfolio_risk = await risk_manager.check_portfolio_risk() if portfolio_risk: print(f" Total Portfolio Risk: {portfolio_risk['total_risk_percentage']:.2f}%") print(f" Risk Limit: {portfolio_risk['max_portfolio_risk_percentage']:.2f}%") print(f" Within Limits: {'✅' if portfolio_risk['within_limits'] else '❌'}") if portfolio_risk["position_details"]: print("\n Position Details:") for pos in portfolio_risk["position_details"]: print(f" {pos['instrument']}: {pos['units']} units, P/L: {pos['unrealized_pl']:.2f}, Risk: {pos['risk_percentage']:.2f}%") else: print("❌ No account connection - cannot assess risk")

Position Sizing Calculation¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        risk_manager = RiskManager(client, account_id)

        # Example position sizing calculation
        instrument = "EUR_USD"
        entry_price = Decimal("1.10000")
        stop_loss = Decimal("1.09500")  # 50 pip stop loss

        print(f"🧮 Position Sizing for {instrument}:")
        print(f"  Entry Price: {entry_price}")
        print(f"  Stop Loss: {stop_loss}")
        print(f"  Risk per Trade: {risk_manager.max_risk_per_trade * 100}%")

        optimal_size = await risk_manager.calculate_position_size(instrument, entry_price, stop_loss)

        print(f"\n✅ Optimal position size: {optimal_size} units")
else:
    print("❌ No account connection - cannot calculate position size")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: risk_manager = RiskManager(client, account_id) # Example position sizing calculation instrument = "EUR_USD" entry_price = Decimal("1.10000") stop_loss = Decimal("1.09500") # 50 pip stop loss print(f"🧮 Position Sizing for {instrument}:") print(f" Entry Price: {entry_price}") print(f" Stop Loss: {stop_loss}") print(f" Risk per Trade: {risk_manager.max_risk_per_trade * 100}%") optimal_size = await risk_manager.calculate_position_size(instrument, entry_price, stop_loss) print(f"\n✅ Optimal position size: {optimal_size} units") else: print("❌ No account connection - cannot calculate position size")

Trade Validation Example¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        risk_manager = RiskManager(client, account_id)

        # Example trade validation
        trade_params = {"instrument": "GBP_USD", "units": 5000, "entry_price": Decimal("1.25000"), "stop_loss": Decimal("1.24500"), "take_profit": Decimal("1.26000")}

        print("🔍 Validating Trade:")
        print(f"  Instrument: {trade_params['instrument']}")
        print(f"  Units: {trade_params['units']}")
        print(f"  Entry: {trade_params['entry_price']}")
        print(f"  Stop Loss: {trade_params['stop_loss']}")
        print(f"  Take Profit: {trade_params['take_profit']}")

        validation = await risk_manager.validate_trade(**trade_params)

        print("\n📋 Validation Result:")
        print(f"  Approved: {'✅' if validation['approved'] else '❌'}")

        if validation["errors"]:
            print("  🚨 Errors:")
            for error in validation["errors"]:
                print(f"    - {error}")

        if validation["warnings"]:
            print("  ⚠️ Warnings:")
            for warning in validation["warnings"]:
                print(f"    - {warning}")

        if validation["adjustments"]:
            print("  💡 Suggested Adjustments:")
            for key, value in validation["adjustments"].items():
                print(f"    - {key}: {value}")
else:
    print("❌ No account connection - cannot validate trade")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: risk_manager = RiskManager(client, account_id) # Example trade validation trade_params = {"instrument": "GBP_USD", "units": 5000, "entry_price": Decimal("1.25000"), "stop_loss": Decimal("1.24500"), "take_profit": Decimal("1.26000")} print("🔍 Validating Trade:") print(f" Instrument: {trade_params['instrument']}") print(f" Units: {trade_params['units']}") print(f" Entry: {trade_params['entry_price']}") print(f" Stop Loss: {trade_params['stop_loss']}") print(f" Take Profit: {trade_params['take_profit']}") validation = await risk_manager.validate_trade(**trade_params) print("\n📋 Validation Result:") print(f" Approved: {'✅' if validation['approved'] else '❌'}") if validation["errors"]: print(" 🚨 Errors:") for error in validation["errors"]: print(f" - {error}") if validation["warnings"]: print(" ⚠️ Warnings:") for warning in validation["warnings"]: print(f" - {warning}") if validation["adjustments"]: print(" 💡 Suggested Adjustments:") for key, value in validation["adjustments"].items(): print(f" - {key}: {value}") else: print("❌ No account connection - cannot validate trade")

Bracket Order Example¶

In [ ]:

if account_id:
    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        advanced_orders = AdvancedOrders(client, account_id)

        print("🎯 Bracket Order Example (Demo - Uncomment to execute):")
        print("  Instrument: EUR_USD")
        print("  Units: 1000")
        print("  Stop Loss: 20 pips")
        print("  Take Profit: 40 pips")

        # Uncomment the lines below to place a real bracket order
        # bracket_result = await advanced_orders.place_bracket_order(
        #     instrument="EUR_USD",
        #     units=1000,
        #     stop_loss_pips=20,
        #     take_profit_pips=40
        # )
        #
        # if bracket_result['success']:
        #     print(f"✅ Bracket order placed successfully!")
        #     print(f"  Trade ID: {bracket_result['trade_id']}")
        #     print(f"  Fill Price: {bracket_result['fill_price']}")
        #     print(f"  Stop Loss: {bracket_result['stop_loss']}")
        #     print(f"  Take Profit: {bracket_result['take_profit']}")
        # else:
        #     print(f"❌ Bracket order failed: {bracket_result['error']}")

        print("\n💡 Uncomment the code above to place a real bracket order")
else:
    print("❌ No account connection - cannot place bracket order")

if account_id: async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: advanced_orders = AdvancedOrders(client, account_id) print("🎯 Bracket Order Example (Demo - Uncomment to execute):") print(" Instrument: EUR_USD") print(" Units: 1000") print(" Stop Loss: 20 pips") print(" Take Profit: 40 pips") # Uncomment the lines below to place a real bracket order # bracket_result = await advanced_orders.place_bracket_order( # instrument="EUR_USD", # units=1000, # stop_loss_pips=20, # take_profit_pips=40 # ) # # if bracket_result['success']: # print(f"✅ Bracket order placed successfully!") # print(f" Trade ID: {bracket_result['trade_id']}") # print(f" Fill Price: {bracket_result['fill_price']}") # print(f" Stop Loss: {bracket_result['stop_loss']}") # print(f" Take Profit: {bracket_result['take_profit']}") # else: # print(f"❌ Bracket order failed: {bracket_result['error']}") print("\n💡 Uncomment the code above to place a real bracket order") else: print("❌ No account connection - cannot place bracket order")

Risk Monitoring Dashboard¶

In [ ]:

async def risk_dashboard(account_id: str):
    """Create a comprehensive risk monitoring dashboard."""

    if not account_id:
        print("❌ No account connection")
        return

    async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client:
        risk_manager = RiskManager(client, account_id)

        print("🏛️ RISK MANAGEMENT DASHBOARD")
        print("=" * 50)

        # Account overview
        account_info = await risk_manager.get_account_info()
        if account_info:
            print("\n💰 Account Overview:")
            print(f"  Balance: {account_info['balance']:,.2f} {account_info['currency']}")
            print(f"  Equity (NAV): {account_info['nav']:,.2f} {account_info['currency']}")
            print(f"  Unrealized P/L: {account_info['unrealized_pl']:,.2f} {account_info['currency']}")

            # Calculate key ratios
            equity_ratio = (account_info["nav"] / account_info["balance"]) * 100
            margin_ratio = (account_info["margin_used"] / account_info["nav"]) * 100 if account_info["nav"] > 0 else 0

            print(f"  Equity Ratio: {equity_ratio:.2f}%")
            print(f"  Margin Usage: {margin_ratio:.2f}%")

        # Risk limits check
        print("\n🛡️ Risk Limits:")
        print(f"  Max Risk per Trade: {risk_manager.max_risk_per_trade * 100:.1f}%")
        print(f"  Max Daily Loss: {risk_manager.max_daily_loss * 100:.1f}%")
        print(f"  Max Portfolio Risk: {risk_manager.max_portfolio_risk * 100:.1f}%")
        print(f"  Max Positions: {risk_manager.max_positions}")

        # Daily loss check
        daily_ok = await risk_manager.check_daily_loss_limit()
        print(f"  Daily Loss Status: {'✅ OK' if daily_ok else '🚨 EXCEEDED'}")

        # Portfolio risk assessment
        portfolio_risk = await risk_manager.check_portfolio_risk()
        if portfolio_risk:
            print("\n📊 Portfolio Risk:")
            print(f"  Current Risk: {portfolio_risk['total_risk_percentage']:.2f}%")
            print(f"  Risk Status: {'✅ OK' if portfolio_risk['within_limits'] else '⚠️ HIGH'}")
            print(f"  Open Positions: {portfolio_risk['position_count']}/{portfolio_risk['max_positions']}")

            if portfolio_risk["position_details"]:
                print("\n📈 Active Positions:")
                for i, pos in enumerate(portfolio_risk["position_details"], 1):
                    status = "🟢" if pos["unrealized_pl"] >= 0 else "🔴"
                    print(f"  {i}. {pos['instrument']}: {pos['units']} units, P/L: {pos['unrealized_pl']:+.2f}, Risk: {pos['risk_percentage']:.1f}% {status}")

        print("\n📋 Risk Assessment Summary:")
        risk_score = 0

        if daily_ok:
            risk_score += 25
            print("  ✅ Daily loss within limits (+25 points)")
        else:
            print("  🚨 Daily loss exceeded (0 points)")

        if portfolio_risk and portfolio_risk["within_limits"]:
            risk_score += 25
            print("  ✅ Portfolio risk acceptable (+25 points)")
        else:
            print("  ⚠️ Portfolio risk elevated (+10 points)")
            risk_score += 10

        if account_info and equity_ratio >= 95:
            risk_score += 25
            print("  ✅ Strong equity position (+25 points)")
        elif account_info and equity_ratio >= 90:
            risk_score += 15
            print("  ⚠️ Moderate equity position (+15 points)")
        else:
            print("  🚨 Weak equity position (0 points)")

        if account_info and margin_ratio <= 50:
            risk_score += 25
            print("  ✅ Conservative margin usage (+25 points)")
        elif account_info and margin_ratio <= 75:
            risk_score += 15
            print("  ⚠️ Moderate margin usage (+15 points)")
        else:
            print("  🚨 High margin usage (0 points)")

        print(f"\n🎯 Overall Risk Score: {risk_score}/100")

        if risk_score >= 90:
            print("  🟢 EXCELLENT - Low risk, safe to trade")
        elif risk_score >= 70:
            print("  🟡 GOOD - Moderate risk, trade with caution")
        elif risk_score >= 50:
            print("  🟠 FAIR - Elevated risk, reduce position sizes")
        else:
            print("  🔴 POOR - High risk, consider closing positions")


# Run the risk dashboard
await risk_dashboard(account_id)

async def risk_dashboard(account_id: str): """Create a comprehensive risk monitoring dashboard.""" if not account_id: print("❌ No account connection") return async with AsyncClient(token=TOKEN, account_id=ACCOUNT_ID, environment=ENVIRONMENT) as client: risk_manager = RiskManager(client, account_id) print("🏛️ RISK MANAGEMENT DASHBOARD") print("=" * 50) # Account overview account_info = await risk_manager.get_account_info() if account_info: print("\n💰 Account Overview:") print(f" Balance: {account_info['balance']:,.2f} {account_info['currency']}") print(f" Equity (NAV): {account_info['nav']:,.2f} {account_info['currency']}") print(f" Unrealized P/L: {account_info['unrealized_pl']:,.2f} {account_info['currency']}") # Calculate key ratios equity_ratio = (account_info["nav"] / account_info["balance"]) * 100 margin_ratio = (account_info["margin_used"] / account_info["nav"]) * 100 if account_info["nav"] > 0 else 0 print(f" Equity Ratio: {equity_ratio:.2f}%") print(f" Margin Usage: {margin_ratio:.2f}%") # Risk limits check print("\n🛡️ Risk Limits:") print(f" Max Risk per Trade: {risk_manager.max_risk_per_trade * 100:.1f}%") print(f" Max Daily Loss: {risk_manager.max_daily_loss * 100:.1f}%") print(f" Max Portfolio Risk: {risk_manager.max_portfolio_risk * 100:.1f}%") print(f" Max Positions: {risk_manager.max_positions}") # Daily loss check daily_ok = await risk_manager.check_daily_loss_limit() print(f" Daily Loss Status: {'✅ OK' if daily_ok else '🚨 EXCEEDED'}") # Portfolio risk assessment portfolio_risk = await risk_manager.check_portfolio_risk() if portfolio_risk: print("\n📊 Portfolio Risk:") print(f" Current Risk: {portfolio_risk['total_risk_percentage']:.2f}%") print(f" Risk Status: {'✅ OK' if portfolio_risk['within_limits'] else '⚠️ HIGH'}") print(f" Open Positions: {portfolio_risk['position_count']}/{portfolio_risk['max_positions']}") if portfolio_risk["position_details"]: print("\n📈 Active Positions:") for i, pos in enumerate(portfolio_risk["position_details"], 1): status = "🟢" if pos["unrealized_pl"] >= 0 else "🔴" print(f" {i}. {pos['instrument']}: {pos['units']} units, P/L: {pos['unrealized_pl']:+.2f}, Risk: {pos['risk_percentage']:.1f}% {status}") print("\n📋 Risk Assessment Summary:") risk_score = 0 if daily_ok: risk_score += 25 print(" ✅ Daily loss within limits (+25 points)") else: print(" 🚨 Daily loss exceeded (0 points)") if portfolio_risk and portfolio_risk["within_limits"]: risk_score += 25 print(" ✅ Portfolio risk acceptable (+25 points)") else: print(" ⚠️ Portfolio risk elevated (+10 points)") risk_score += 10 if account_info and equity_ratio >= 95: risk_score += 25 print(" ✅ Strong equity position (+25 points)") elif account_info and equity_ratio >= 90: risk_score += 15 print(" ⚠️ Moderate equity position (+15 points)") else: print(" 🚨 Weak equity position (0 points)") if account_info and margin_ratio <= 50: risk_score += 25 print(" ✅ Conservative margin usage (+25 points)") elif account_info and margin_ratio <= 75: risk_score += 15 print(" ⚠️ Moderate margin usage (+15 points)") else: print(" 🚨 High margin usage (0 points)") print(f"\n🎯 Overall Risk Score: {risk_score}/100") if risk_score >= 90: print(" 🟢 EXCELLENT - Low risk, safe to trade") elif risk_score >= 70: print(" 🟡 GOOD - Moderate risk, trade with caution") elif risk_score >= 50: print(" 🟠 FAIR - Elevated risk, reduce position sizes") else: print(" 🔴 POOR - High risk, consider closing positions") # Run the risk dashboard await risk_dashboard(account_id)

Interpreting the results¶

Check the printed inputs, timestamps and handled errors before interpreting output. A completed cell can still report an API failure. Example calculations and simulations do not establish a profitable strategy, a guaranteed loss cap or readiness for unattended execution.

For further work, test boundary cases, currency conversion, incomplete data and recovery after a disconnect. Keep signal evaluation separate from order submission and reconcile any account changes.

• SDK lifecycle and data types
• Error handling
• Streaming behavior